关于java:rpush多平台统一消息推送系统

https://gitee.com/shuangmulin/rpush

一个接口触达多平台（包含微信公众号、企业微信、钉钉、邮箱等任何想的到平台，都能一个接口一次推送；极简的代码调用，极大缩小业务方音讯推送的代码量）；同时提供基于netty和websocket的即时通讯实现，实现单聊、群聊等性能。开箱即用，采纳SpringCloud微服务架构，扩大简略且没有单点问题。致力于包揽所有和音讯推送无关的技术开发工作，节俭开发资源。

一个接口触达多平台，反对一个接口多平台同时发送
音讯平台逻辑与业务逻辑的解耦，业务方不须要关怀各个平台的对接实现，只须要关怀：要用哪些平台发、要发给对应平台的哪些人、要发什么内容
极强的扩展性，要新增一个音讯平台的反对，实践只须要新增几个类就能实现，且不须要写任何前端代码即可取得该平台对应的ui交互（包含：配置交互、接管人保护、web手动音讯发送交互等）。
当然反对web端手动发送音讯
当然也反对定时工作
音讯计划预设置
提供即时通讯实现，且反对服务器横向扩大
接管人导入
接管人按分组划分
消息日志
...

在线体验

http://159.75.121.163/
admin admin

目前反对的音讯类型

邮箱
企业微信-利用音讯
- 文本音讯
- 图片音讯
- 视频音讯
- 文本音讯
- 文本卡片音讯
- 图文音讯
- Markdown音讯
企业微信-群机器人
- 文本音讯
- 图片音讯
- 图文音讯
- Markdown音讯
微信公众号
- 文本音讯
- 图文音讯
- 模板音讯
钉钉-工作告诉
- 文本
- Markdown
- 链接音讯
- 卡片音讯
- OA音讯
钉钉-群机器人
- 文本
- Markdown
- 链接音讯
- 卡片音讯
- FeedCard

Rpush的架构决定了扩大一个音讯平台的音讯类型会非常简单，所以如果要扩大一个音讯平台，大部分工夫都会花在查找该平台的对接文档上。后续会在工作之余加上其它的平台或音讯类型。当然，欢送参加扩大（扩大一个音讯平台的音讯类型，只须要几个java类即可，不须要写任何前端代码，即可取得包含ui交互内的所有性能）。

本地如何疾速部署一个简略的体验版

本地数据库执行sql/表构造.sql初始化数据库，初始化好数据库之后，到rpush-route/src/main/resources/application.yml把数据库连贯配置改成本人的数据库。而后别离启动以下3个类：

com.regent.rpush.eureka.EurekaServerApplication
com.regent.rpush.zuul.App
com.regent.rpush.route.RouteApplication

启动实现之后，浏览器关上http://localhost:8124就能显示界面并失去一个残缺的音讯散发性能

成果展现

单个音讯类型发送示例

web端多平台发送示例

postman多平台发送示例

用代码发消息

秉持”业务服务只负责发消息“的解耦准则，业务服务在须要发消息的时候，代码应该越简略越好。所以，Rpush的发消息的sdk，一种音讯只须要一行代码，有几种音讯就有几行代码。比方这样：

/** * @author shuangmulin * @since 2021/6/8/008 11:37 **/public class RpushSenderTest {    /**     * 要发送的内容     */    public static final String content = "您的会议室曾经预约 \n" +            ">**事项详情** \n" +            ">事　项：散会\n" +            ">组织者：@miglioguan \n" +            ">参与者：@miglioguan、@kunliu、@jamdeezhou、@kanexiong、@kisonwang \n" +            "> \n" +            ">会议室：广州TIT 1楼 301\n" +            ">日　期：2021年5月18日\n" +            ">时　间：上午9:00-11:00\n" +            "> \n" +            ">请准时加入会议。 \n" +            "> \n" +            ">如需批改会议信息，请点击：[批改会议信息](https://work.weixin.qq.com)";    public static void main(String[] args) {        // 企业微信-markdown音讯        MarkdownMessageDTO markdown = RpushMessage.WECHAT_WORK_AGENT_MARKDOWN().content(content).receiverIds(Collections.singletonList("ZhongBaoLin")).build();        // 企业微信-群机器人音讯        TextMessageDTO text = RpushMessage.WECHAT_WORK_ROBOT_TEXT().content(content).receiverIds(Collections.singletonList("ZhongBaoLin")).build();        // 邮箱        EmailMessageDTO email = RpushMessage.EMAIL().title("会议告诉").content(content).build();        RpushService.instance("baolin", "666666").sendMessage(markdown, text, email); // 填上账号密码，运行即可    }}

以上代码，一次发送了三种不同平台的不同音讯类型，全副代码加起来也只须要四五行代码而已。要取得以上成果，只须要maven援用rpush的sdk模块即可：

<project>    <!-- 设置 jitpack.io 仓库 -->    <repositories>        <repository>            <id>jitpack.io</id>            <url>https://jitpack.io</url>        </repository>    </repositories>    <dependencies>        <!-- 增加rpush-sdk依赖 -->        <dependency>            <groupId>com.github.shuangmulin.rpush</groupId>            <artifactId>rpush-sdk</artifactId>            <version>v1.0.2</version>        </dependency>    </dependencies></project>

即时通讯

Rpush对即时通讯的实现形式比拟容纳
，即对具体的连贯实现做理解耦，不局限于某一种连贯形式，能够netty，能够websocket，能够comet，当然也能够用原始的bio来做。这里展现websocke的网页端和netty实现的命令行客户端之间相互单聊和群聊的成果（该示例的相干代码：客户端示例代码地址）：

一些比拟外围的扩大点

1. 可自在扩大的音讯平台和音讯类型

在Rpush的设计里，音讯被归类为“音讯平台”和“音讯类型”，别离对应如下两个枚举：

/** * 音讯平台枚举 **/public enum MessagePlatformEnum {    EMAIL(EmailConfig.class, "邮箱", "", "^[_a-z0-9-]+(\\.[_a-z0-9-]+)*@[a-z0-9-]+(\\.[a-z0-9-]+)*(\\.[a-z]{2,})$", true),    WECHAT_WORK_AGENT(WechatWorkAgentConfig.class, "企业微信-利用音讯", "", "", true),    WECHAT_WORK_ROBOT(WechatWorkRobotConfig.class, "企业微信-群机器人", "", "", true),    WECHAT_OFFICIAL_ACCOUNT(WechatOfficialAccountConfig.class, "微信公众号", "", "", true),    DING_TALK_CORP(DingTalkCorpConfig.class, "钉钉-工作告诉", "", "", true),    RPUSH_SERVER(EmptyConfig.class, "rpush服务", "", "", true);}/** * 音讯类型枚举 **/public enum MessageType {    EMAIL("一般邮件 ", MessagePlatformEnum.EMAIL),    RPUSH_SERVER("文本", MessagePlatformEnum.RPUSH_SERVER),    // ================================企业微信-利用====================================    WECHAT_WORK_AGENT_TEXT("文本", MessagePlatformEnum.WECHAT_WORK_AGENT),    WECHAT_WORK_AGENT_IMAGE("图片", MessagePlatformEnum.WECHAT_WORK_AGENT),    WECHAT_WORK_AGENT_VIDEO("视频", MessagePlatformEnum.WECHAT_WORK_AGENT),    WECHAT_WORK_AGENT_FILE("文件", MessagePlatformEnum.WECHAT_WORK_AGENT),    WECHAT_WORK_AGENT_TEXTCARD("文本卡片", MessagePlatformEnum.WECHAT_WORK_AGENT),    WECHAT_WORK_AGENT_NEWS("图文音讯", MessagePlatformEnum.WECHAT_WORK_AGENT),    WECHAT_WORK_AGENT_MARKDOWN("Markdown", MessagePlatformEnum.WECHAT_WORK_AGENT),    // ================================企业微信-群机器人====================================    WECHAT_WORK_ROBOT_TEXT("文本", MessagePlatformEnum.WECHAT_WORK_ROBOT),    WECHAT_WORK_ROBOT_IMAGE("图片", MessagePlatformEnum.WECHAT_WORK_ROBOT),    WECHAT_WORK_ROBOT_NEWS("图文音讯", MessagePlatformEnum.WECHAT_WORK_ROBOT),    WECHAT_WORK_ROBOT_MARKDOWN("Markdown", MessagePlatformEnum.WECHAT_WORK_ROBOT),    // ================================微信公众号====================================    WECHAT_OFFICIAL_ACCOUNT_TEXT("文本", MessagePlatformEnum.WECHAT_OFFICIAL_ACCOUNT),    WECHAT_OFFICIAL_ACCOUNT_NEWS("图文音讯", MessagePlatformEnum.WECHAT_OFFICIAL_ACCOUNT),    WECHAT_OFFICIAL_ACCOUNT_TEMPLATE("模板音讯", MessagePlatformEnum.WECHAT_OFFICIAL_ACCOUNT),    // ================================钉钉-工作告诉====================================    DING_TALK_COPR_TEXT("文本", MessagePlatformEnum.DING_TALK_CORP),    DING_TALK_COPR_MARKDOWN("Markdown", MessagePlatformEnum.DING_TALK_CORP),    DING_TALK_COPR_LINK("链接音讯", MessagePlatformEnum.DING_TALK_CORP),    DING_TALK_COPR_ACTION_CARD_SINGLE("卡片-单按钮", MessagePlatformEnum.DING_TALK_CORP),    DING_TALK_COPR_ACTION_CARD_MULTI("卡片-多按钮", MessagePlatformEnum.DING_TALK_CORP),    DING_TALK_COPR_OA("OA音讯", MessagePlatformEnum.DING_TALK_CORP),    ;}

这里拿“企业微信-利用的文本类型”的音讯举例。假如当初要在Rpush实现这个类型的音讯，步骤如下：

定义企业微信的配置类，如下：

/** * 企业微信配置 **/@EqualsAndHashCode(callSuper = true)@Data@NoArgsConstructor@AllArgsConstructor@Builderpublic class WechatWorkAgentConfig extends Config {    private static final long serialVersionUID = -9206902816158196669L;    @ConfigValue(value = "企业ID", description = "在此页面查看：https://work.weixin.qq.com/wework_admin/frame#profile")    private String corpId;    @ConfigValue(value = "利用Secret")    private String secret;    @ConfigValue(value = "利用agentId")    private Integer agentId;}

外面的字段就按对应平台须要的字段去定义就行，比方这里的企业微信就只有三个字段须要配置。而每个字段上的@ConfigValue
注解，是用来主动生成页面的，也就是说，只须要打上这个注解，就能够主动在页面上生成对应的增删改查的界面和交互（无需写一行前端代码）。

在MessagePlatformEnum和MessageType
里加上对应的枚举，即WECHAT_WORK_AGENT(WechatWorkAgentConfig.class, "企业微信-利用音讯", "", "", true)
和WECHAT_WORK_AGENT_TEXT("文本", MessagePlatformEnum.WECHAT_WORK_AGENT),。这里要留神下平台枚举的第一个参数就是第一步定义的配置类的Class。
定义企业微信-利用-文本音讯的参数，如下：

/** * 企业微信音讯发送DTO **/@EqualsAndHashCode(callSuper = true)@Data@SuperBuilder@NoArgsConstructor@AllArgsConstructorpublic class TextMessageDTO extends BaseMessage {    private static final long serialVersionUID = -3289428483627765265L;    /**     * 接管人分组列表     */    @SchemeValue(type = SchemeValueType.RECEIVER_GROUP)    private List<Long> receiverGroupIds;    /**     * 接管人列表     */    @SchemeValue(type = SchemeValueType.RECEIVER)    private List<String> receiverIds;    @SchemeValue(description = "PartyID列表，非必填，多个接受者用‘|’分隔。当touser为@all时疏忽本参数")    private String toParty;    @SchemeValue(description = "TagID列表，非必填，多个接受者用‘|’分隔。当touser为@all时疏忽本参数")    private String toTag;    @SchemeValue(type = SchemeValueType.TEXTAREA, description = "请输出内容...")    private String content;}

同样的，外面的字段依据该音讯类型须要的字段去定义就行。比方企业微信-利用-文本音讯就只须要一个content内容字段以及接管人相干的字段。这里波及到的@SchemeValue
注解，同样也是用来主动生成页面交互的，即只须要打上这个注解，就能主动在发消息页面生成对应的ui和交互。同时能够应用com.regent.rpush.route.utils.sdk.SdkGenerator类主动生成sdk代码。

实现com.regent.rpush.route.handler.MessageHandler接口，正式写发消息的代码。

/** * 企业微信文本音讯handler **/@Componentpublic class AgentTextMessageHandler extends MessageHandler<TextMessageDTO> {    @Override    public MessageType messageType() {        return MessageType.WECHAT_WORK_AGENT_TEXT;    }    @Override    public void handle(TextMessageDTO param) {        // 具体的发消息代码    }}

这里有以下须要关怀的点：

接口上的泛型填第3步定义的类
实现messageType办法，返回以后类要解决的音讯类型
实现handle办法，写发消息的代码，外面的参数是主动解析到这个办法的，间接应用即可

到这里，就不须要多做任何其它的事了。也就是说，做完以上四个步骤，就曾经实现了一个音讯类型的扩大。事做的少，取得的性能并不少：

主动取得对应平台配置的增删改交互和ui
主动取得对应平台接管人和接管人分组的增删改交互和ui（包含导入性能）
主动取得该音讯类型手动发送音讯的交互和ui
执行com.regent.rpush.route.utils.sdk.SdkGenerator，主动实现该音讯类型的sdk代码
主动取得该音讯类型的定时工作的增删改交互和ui

而且减少一个音讯类型，不会对业务服务之前正在应用的音讯类型有任何影响，是纯正的叠加”能力“。

2. 可自在扩大的即时通讯实现

在rpush的架构里，投递一个音讯的流程大抵能够概括为：调用对立的接口向路由服务投递音讯-》路由服务查出音讯指标所在的服务器地址-》路由服务向对应的服务器传递音讯-》对应的服务找到对应的会话发送音讯。
这里的扩大点在最初一步，即用户和服务器的会话保护。要实现服务端向客户端推送音讯，会有比拟多的解决方案，比方用netty起一个nio服务器，客户端去连netty服务器或者服务端用socketio提供websocket实现，客户端按websocket的形式连服务器或者用comet实现长连贯让客户端连等等。

这里做成扩大点的一个比拟重要的思考点就是要实现不同端之间的音讯通信，比方下面例子里的命令行和网页之间的聊天，或者实现挪动端和网页之间的聊天。

RpushClient

不论是什么技术实现的服务端推送，都会有一个“客户端”性质的类，比方netty会有Channel，socketio提供的websocket会有SocketIOClient
。而对于rpush来说，只关怀它们的一个共有的能力：音讯投递。即RpushClient接口：

/** * 客户端 **/public interface RpushClient {    /**     * 推送音讯     */    void pushMessage(NormalMessageDTO message);    void close();}

只有实现了这个接口的类，不论是什么技术的实现，都被认为是rpush的客户端。也就是说，netty也好，websocket也好，只有提供给rpush这个接口的能力即可，从而达到解耦具体实现的目标。
目前rpush曾经做了netty和socketio两个实现，别离对应com.regent.rpush.server.socket.nio.NioSocketChannelClient
和com.regent.rpush.server.socket.websocket.WebSocketClient两个类。

netty客户端sdk

rpush提供了netty对应的客户端的sdk，我的项目依赖rpush-client即可，应用也非常简单，只须要几行代码即可。

public class Main {    public static void main(String[] args) {        RpushClient rpushClient = new RpushClient(servicePath, registrationId); // 填上rpush服务地址和id        rpushClient.addMsgProcessor(new PingIgnoreMsgProcessor()); // 疏忽心跳音讯        rpushClient.start(); // 向服务端发动连贯        rpushClient.addMsgProcessor(msg -> {            // 解决接管到的音讯            return false;        });    }}

对于架构

rpush目前次要提供两大性能，一个是音讯散发，另一个是即时通讯性能。音讯散发由路由服务rpush-route提供，即时通讯的长连贯保护由socket服务rpush-server服务提供。

1. 可自在集群的路由服务

为了保障音讯投递的统一性以及解耦音讯散发和即时通讯之间的关系，路由服务只做一件事，即负责将音讯散发到各个平台，也就是说rpush提供的即时通讯性能，对路由服务来说和其余第三方平台没什么区别，都被视为一个平台。
所以在架构层面，如果只须要用到音讯散发的性能，就不须要部署rpush-server
服务，只须要部署eureka、zuul和路由服务即可。zuul作为零碎对外的入口，隔离掉了路由服务器和用户端，同时路由服务又是无状态的，这样就使得路由服务能够依据理论业务状况自在集群，即想加一台路由服务就加，想减一台就减。

2. 可自在集群的socket服务

rpush-server作为socket服务，次要性能就是保护客户端的长连贯。这个服务的承载能力间接决定了即时通讯性能一次能够在线多少用户，所以这个服务毫无疑问必须要是可集群部署的。
假如部署5台socket服务，都失常配置eureka为注册核心。为了实现socket服务的集群，一个客户端连贯rpush服务的流程为：

客户端问路由服务要一个可用的socket服务器ip和端口
路由服务通过适合的负载平衡算法失去一个可用的socket服务器ip和端口并返回给客户端
客户端向拿到的socket服务发动长连贯
连贯胜利后，对应的socket服务器保护服务级别的session信息，而后向路由服务汇报该客户端，路由服务保留该客户端和socket服务的对应关系
客户端与对应的socket服务放弃肯定频率的心跳，并在心跳失败断定连贯断开后从新发动以上流程，直到再次连贯胜利

客户端基于以上步骤上线之后，其余客户端向该客户端投递音讯的流程为：

客户端申请路由服务提供的音讯投递接口（这个就是后面说到的音讯投递接口，因为socket服务保护的长连贯对路由服务来说和其余第三方平台没什么区别，所以音讯投递的形式也是一样的）
路由服务实现的socket服务音讯处理器（com.regent.rpush.route.handler.RpushMessageHandler），依据音讯上的指标客户端id找到对应的socket服务，并向该服务投递音讯
socket服务从本人保护的session里找到指标客户端，最终实现音讯投递

实现以上流程之后，socket服务就能够做到自在集群了。

下面说的流程偏理论化，有几个技术实现点这里做一下具体阐明：

路由服务如何通过适合的负载平衡算法失去一个可用的socket服务器ip和端口？

实现的伎俩其实十分的简略暴力。首先由socket服务提供一个查问本机ip和端口的接口，路由服务间接通过ribbon去申请这个接口，而后自定义一个负载平衡规定类，来实现socket服务的抉择：

/** * 路由->Socket服务端申请的实例抉择 */public class ServerBalancer extends ZoneAvoidanceRule {    @Override    public Server choose(Object o) {        // ...        // 用默认的负载平衡算法选出一个可用的socket服务（这里的算法能够依据理论业务更改）        return super.choose(o);        // ...    }}

在配置文件里配置这个“规定类”：

rpush-server:  ribbon:    NFLoadBalancerRuleClassName: com.regent.rpush.route.loadbalancer.ServerBalancer

路由服务在向socket服务申请的时候会”通过“这个”规定类“，而后由这个“规定类”来选出一个可用的socket服务。最终socket服务的端口和ip信息，也是由选中的socket服务通过这次申请返回给路由服务的。
当然这个规定类不是只做这一件事，还有一个问题也须要这个类来实现。

音讯投递的时候，路由服务如何依据音讯上的指标客户端id找到对应的socket服务？

首先，在客户端与某一个socket服务连接成之后，客户端与socket服务之间的关系须要保存起来（mysql或redis）。而后新增一个feign的申请拦截器（com.regent.rpush.route.loadbalancer.MessageRequestInterceptor
）：

@Componentpublic class MessageRequestInterceptor implements RequestInterceptor {    /**     * 寄存本次音讯投递的指标socket服务id     */    static final ThreadLocal<String> SERVER_ID = new ThreadLocal<>();    @Autowired    private IRpushServerOnlineService rpushServerOnlineService;    @SuppressWarnings("MismatchedQueryAndUpdateOfCollection")    @Override    public void apply(RequestTemplate requestTemplate) {        String url = requestTemplate.url();        String method = requestTemplate.method();        if (!"/push".equals(url) || !"POST".equals(method)) {            // 只解决音讯投递接口            return;        }        // 如果是音讯推送，须要给接收端连贯的服务端投放音讯，在服务端集群的状况下，要找到对应的服务端        String body = new String(requestTemplate.body());        JSONObject jsonObject = new JSONObject(body);        String sendTo = jsonObject.getStr("sendTo"); // 拿到指标客户端的id        String serverId = ""; // 从redis或mysql查到该客户端对应的socket服务id        SERVER_ID.set(serverId); // 增加到以后线程里    }}

这个“拦挡类”配合下面的”规定类“，就能在路由服务向socket服务传递音讯时精确的找到对应的socket服务。残缺的”规定类“：

/** * 路由->Socket服务端申请的实例抉择 */public class ServerBalancer extends ZoneAvoidanceRule {    @Override    public Server choose(Object o) {        try {            // 从拦挡类里看有没有指定服务端实例            String serverId = MessageRequestInterceptor.SERVER_ID.get();            if (StringUtils.isEmpty(serverId)) {                // 如果没有指定服务端实例，用默认的负载平衡算法                return super.choose(o);            }            // 如果指定了服务端实例，阐明是消息传递，用指定好的实例向socket服务发申请            List<Server> servers = getLoadBalancer().getAllServers();            for (Server server : servers) {                if (StringUtils.equals(server.getId(), serverId)) {                    return server;                }            }            throw new IllegalArgumentException("没有可用的RPUSH_SERVER实例");        } finally {            MessageRequestInterceptor.SERVER_ID.remove();        }    }}

而且有了这两个类，路由服务向socket服务传递音讯的代码也会十分的”洁净“：

@Componentpublic class RpushMessageHandler extends MessageHandler<RpushMessageDTO> {    // ...    @Override    public void handle(RpushMessageDTO param) {        List<String> sendTos = param.getReceiverIds();        for (String sendTo : sendTos) {            // ...            messagePushService.push(build); // 路由服务间接调用接口申请即可，”规定类“和”拦挡类“屏蔽掉了其它逻辑，所以这里不须要关怀会不会发给谬误socket服务        }    }}

3. 其它

队列。路由服务外部用Disruptor环形队列做了异步解决，尽可能地让音讯推送接口更快地返回。如果是并发量较高的状况，能够退出kafka，路由服务间接监听kafka的音讯，以此来晋升服务整体性能。
缓存。客户端的上线信息可依据状况做多级缓存。即路由服务外部缓存+redis缓存，当然加的缓存越多，缓存一致性的问题就越简单，须要思考的状况也会更多。redis也是须要依据理论状况来决定是否要集群部署。
监控。可应用Spring Boot Admin做服务状态监控。

用docker-compose疾速部署一个Rpush服务

version: '2'services:  nginx:    image: nginx    container_name: nginx    ports:      - 80:80    volumes:      - /data/nginx/conf/nginx.conf:/etc/nginx/nginx.conf      - /data/nginx/log:/var/log/nginx      - /data/nginx/html:/usr/share/nginx/html  rpush-eureka:    image: shuangmulin/rpush-eureka    container_name: rpush-eureka    ports:      - 8761:8761  rpush-zuul:    image: shuangmulin/rpush-zuul    environment:      - eureka-service-ip=172.16.0.11      - eureka-service-port=8761    container_name: rpush-zuul    ports:      - 8124:8124  rpush-route:    image: shuangmulin/rpush-route    environment:      - eureka-service-ip=localhost      - eureka-service-port=8761      - jdbc.url=jdbc:mysql://localhost:3306/rpush?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=GMT%2B8      - jdbc.username=root      - jdbc.password=123456      - super-admin.username=superadmin      - super-admin.password=superadmin      - jwtSigningKey=fjksadjfklds    container_name: rpush-route    ports:      - 8121:8121  rpush-server:    image: shuangmulin/rpush-server    environment:      - eureka-service-ip=localhost      - eureka-service-port=8761    container_name: rpush-server    ports:      - 8122:8122  rpush-scheduler:    image: shuangmulin/rpush-scheduler    environment:      - eureka-service-ip=localhost      - eureka-service-port=8761      - jdbc.url=jdbc:mysql://localhost:3306/rpush?useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=GMT%2B8      - jdbc.username=root      - jdbc.password=123456      - super-admin.username=superadmin      - super-admin.password=superadmin      - jwtSigningKey=fasdferear    container_name: rpush-scheduler    ports:      - 8123:8123

运行 docker-compose up -d之后，间接拜访8124端口即可