详解在spring boot中消息推送系统设计与实现
推送系统作为通用的组件，存在的价值主要有以下几点
- 会被多个业务项目使用，推送系统独立维护可降低维护成本
- 推送系统一般都是调用三方api进行推送，三方api一般会有调用频率/次数限制，被推送的消息需要走队列来合理调用三方api，控制调用的频率和次数
- 业务无关，一般推送系统设计成不需要关心业务逻辑
核心技术
- 消息队列
- 三方服务api调用
- 安卓app推送
- 苹果app推送
- 微信小程序推送
- 邮件推送
- 钉钉推送
- 短信推送
消息队列选用阿里云提供的rocketmq，官方文档：
推送时序图
右键新窗口打开可以查看高清大图
可以看到消息推送系统接入的第三方服务比较多，三方推送的质量很难统一，就需要考虑消息的推送的重试了
思路流程
为了控制并发，所有的推送都先发到rocketmq队列里，每次推送的个数通过控制队列的消费的客户端的数量来实现
安卓和苹果都可以使用信鸽的推送服务
信鸽推送需要客户端进行集成，客户端sdk参考：
目前信鸽个人开发者仍然是可以申请的，账号建立后，创建andorid和ios项目
记录下这里的 app id和secret key，服务端进行推送时需要这两个参数
推送异常处理，推送异常时，需要重试，重试可以利用消息队列本身的重试机制，也可以自行实现重试逻辑
安卓app推送
官方文档：
代码库： https://github.com/xingepush/xinge-api-java
<!-- 信鸽推送客户端 --> <dependency> <groupid>com.github.xingepush</groupid> <artifactid>xinge</artifactid> <version>1.2.1</version> </dependency>
核心代码如下
map<string, object> messagepayload = new hashmap<string, object>(1);
messagepayload.put("user_id", 1);
messagepayload.put("msg_title", "消息标题");
messagepayload.put("msg_content", "消息内容");
messagepayload.put("msg_type", 1);
messagepayload.put("data", lists.newhashmap("order_id", 1));
xingeapp xinge = new xingeapp(androidaccessid, androidsecretkey);
messageandroid message = new messageandroid();
clickaction action = new clickaction();
action.setactiontype(clickaction.type_activity);
message.setaction(action);
message.setcontent(jsonutil.tojsonstring(messagepayload));
message.settype(messageandroid.type_message);
message.setexpiretime(86400);
message.setcustom(new hashmap<string, object>(1));
jsonobject response = xinge.pushsingledevice("用户的pushtoken", message);
if (response.getint(ret_code) != 0) {
// 推送异常了，进行日志记录等
}
苹果app推送
使用pushy库进行推送
<!-- ios推送客户端 --> <dependency> <groupid>com.turo</groupid> <artifactid>pushy</artifactid> <version>0.13.3</version> </dependency>
map<string, object> aps = new hashmap<string, object>(1);
aps.put("alert", "推送内容");
aps.put("sound", "default");
aps.put("badge", 1);
map<string, object> data = new hashmap<string, object>(1);
data.put("msgtitle", "推送标题");
data.put("msgcontent", "推送内容");
data.put("msgtype", "1");
data.put("userid", "13288888888");
data.put("data", lists.newhashmap("order_id", 1));
map<string, object> messagepayload = new hashmap<string, object>(1);
messagepayload.put("aps", aps);
messagepayload.put("data", data);
apnsclient apnsclient = new apnsclientbuilder()
.setapnsserver(apnsclientbuilder.production_apns_host)
.setclientcredentials(this.getclass().getclassloader().getresourceasstream("推送证书相对resources目录的路径"), "")
.build();
string payload = jsonutil.tojsonstring(messagepayload);
string token = tokenutil.sanitizetokenstring("app用户的pushtoken");
simpleapnspushnotification pushnotification = new simpleapnspushnotification(token, "com.suxiaolin.app1", payload);
pushnotificationfuture<simpleapnspushnotification, pushnotificationresponse<simpleapnspushnotification>>
sendnotificationfuture = apnsclient.sendnotification(pushnotification);
final pushnotificationresponse<simpleapnspushnotification> pushnotificationresponse =
sendnotificationfuture.get();
if (pushnotificationresponse.isaccepted()) {
system.out.println("push notification accepted by apns gateway.");
} else {
system.out.println("notification rejected by the apns gateway: " +
pushnotificationresponse.getrejectionreason());
if (pushnotificationresponse.gettokeninvalidationtimestamp() != null) {
system.out.println("\t…and the token is invalid as of " +
pushnotificationresponse.gettokeninvalidationtimestamp());
}
}
使用信鸽库进行推送
当然也可以使用信鸽提供的ios推送，逻辑和安卓app的推送差不多
ios的信鸽客户端可以和android的客户端共用，不需要引入新的jar包了
jsonobject messagepayload = new jsonobject();
map<string, object> custom = new hashmap<string, object>();
messagepayload.put("title", "推送标题");
messagepayload.put("body", "推送内容");
messagepayload.put("user_id", 1);
messagepayload.put("msg_type", 1);
messagepayload.put("data", lists.newarraylist("orderid", 1));
xingeapp xinge = new xingeapp(iosaccessid, iossecretkey);
messageios message = new messageios();
message.settype(messageios.type_apns_notification);
message.setexpiretime(86400);
message.setalert(content);
message.setbadge(1);
message.setcategory("invite_category");
message.setcustom(messagepayload);
response = xinge.pushsingledevice("app用户的pushtoken", message, xingeapp.iosenv_prod);
if (response.getint(ret_code) != 0) {
// 推送异常了
}
小程序推送
官方文档：
可以看到微信小程序推送接口是普通的post请求
小程序api地址：
http请求，http请求库可以参考： httputil
钉钉推送
官方文档： 代码示例
public static boolean send(string content, string title, set<string> receivers) {
try {
httputil.responsewrap result = httputil.postwrap(ddurl,
"{\n"
+ " \"msgtype\": \"text\",\n"
+ " \"text\": {\"content\":\"" + title + "\r\n" + content + "\n|"
+ receivers.stream().map(r -> "@" + r).collect(collectors.joining(" ")) + "\"},\n"
+ " \"at\": {\n"
+ " \"atmobiles\": [" + receivers.stream().map(r -> "\"" + r + "\"").collect(collectors.joining(",")) + "], \n"
+ " \"isatall\": false\n"
+ " }\n"
+ " }");
return result.getstatuscode() == 200;
} catch (exception e) {
return false;
}
}
完整代码参考dingtalkutil.java
使用http请求就可以请求了
邮件推送
发送邮件可以使用java的javax.mail库，smtp协议发送邮件
<dependency> <groupid>javax.mail</groupid> <artifactid>mail</artifactid> <version>1.4.7</version> </dependency>
示例代码参考： emailsender.java
短信推送
短信服务商众多，邮件一般有统一的smtp协议可以使用，短信没有协议，但是一般使用http发送短信 比如以下的短信服务商
253云通讯：
短信服务- 又拍云:
消息&短信_msgsms_云通信_短信- 华为云:
消息队列的特性
消息队列消费异常后会自动进行重试
一些注意的点
微信小程序每次支付可以生成一个推送码，需要保存到数据库或者缓存里，并且每个码只能推送3条消息
因为消息队列的消费在消息量大的时候具有一定的延时，这就为取消消息推送提供了可能，可以为每条消息生成一个唯一的uuid，取消的时候把这个uuid设计进redis里，推送时检查这个uuid是否在redis里决定推送与否
虽然推送存在不可控制的异常，比如三方推送服务出现了异常，但是也存在调用方传递参数异常，可以推送接口调用的返回值判断是否调用推送系统成功，也可以记录到日志里，这样在调查异常原因时就比较容易
消息队列默认的重试次数，消费时长是无法控制的，可以对消息队列的客户端进行修改支持这个特性，参考： 核心逻辑是先给消息设置一个最大消费次数和消费时长，然后当消息消费次数和消费时长达到阈值时，直接置为成功
ios使用信鸽推送时，需要上传开发证书和生产证书，这两个证书至少需要上传一个
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。