工作七年，对消息推送使用的一些经验和总结

大家好，欢迎来到IT知识分享网。

前言：不管是APP还是WEB端都离不开消息推送，尤其是APP端，push消息，小信箱消息；WEB端的代办消息等。因在项目中多次使用消息推送且也是很多项目必不可少的组成部分，故此总结下供自己参考。

一、什么是消息推送

消息推送（Push）指运营人员通过自己的产品或第三方工具对用户当前网页或移动设备进行的主动消息推送。用户可以在网页上或移动设备锁定屏幕和通知栏看到push消息通知

二、消息推送的种类

从数据模型分：推和拉

从终端分：APP端和WEB端

从实现层面分：短论询、Comet（长轮询）、Flash XMLSocket、SSE、Web-Socket

短轮询：客户端定时轮询发起请求

长轮询：客户端发起请求，等待后端响应并再次发起请求

Flash XMLSocket：

原理示意图：

利用Flash XML Socket实现”服务器推”技术前提：
（1）Flash提供了XMLSocket类，服务器利用Socket向Flash发送数据；
（2）JavaScript和Flash的紧密结合JavaScript和Flash可以相互调用。
优点是实现了socket通信，不再利用无状态的http进行伪推送。但是缺点更明显：
1.客户端必须安装 Flash 播放器；
2.因为 XMLSocket 没有 HTTP 隧道功能，XMLSocket 类不能自动穿过防火墙；
3.因为是使用套接口，需要设置一个通信端口，防火墙、代理服务器也可能对非 HTTP 通道端口进行限制。

SSE：当使用Server-Sent Events（SSE）时，客户端（通常是浏览器）与服务器之间建立一种持久的连接，使服务器能够主动向客户端发送数据。这种单向的、服务器主动推送数据的通信模式使得实时更新的数据能够被实时地传送到客户端，而无需客户端进行轮询请求

SSE的工作原理如下： 

1. 建立连接：客户端通过使用EventSource对象在浏览器中创建一个与服务器的连接。客户端向服务器发送一个HTTP请求，请求的头部包含Accept: text/event-stream，以表明客户端希望接收SSE数据。服务器响应这个请求，并建立一个持久的HTTP连接。
2. 保持连接：服务器保持与客户端的连接打开状态，不断发送数据。这个连接是单向的，只允许服务器向客户端发送数据，客户端不能向服务器发送数据。
3. 服务器发送事件：服务器使用Content-Type: text/event-stream标头来指示响应是SSE数据流。服务器将数据封装在特定的SSE格式中，每个事件都以data:开头，后面是实际的数据内容，以及可选的其他字段，如event:和id:。服务器发送的数据可以是任何文本格式，通常是JSON。
4. 客户端接收事件：客户端通过EventSource对象监听服务器发送的事件。当服务器发送事件时，EventSource对象会触发相应的事件处理程序，开发人员可以在处理程序中获取到事件数据并进行相应的操作。常见的事件是message事件，表示接收到新的消息。
5. 断开连接：当客户端不再需要接收服务器的事件时，可以关闭连接。客户端可以调用EventSource对象的close()方法来显式关闭连接，或者浏览器在页面卸载时会自动关闭连接。

在Spring Boot中，可以使用SseEmitter类来实现SSE：

@RestController public class SSEController { private SseEmitter sseEmitter; @GetMapping("/subscribe") public SseEmitter subscribe() { sseEmitter = new SseEmitter(); return sseEmitter; } @PostMapping("/send-message") public void sendMessage(@RequestBody String message) { try { if (sseEmitter != null) { sseEmitter.send(SseEmitter.event().data(message)); } } catch (IOException e) { e.printStackTrace(); } } }

<script> // 创建一个EventSource对象，指定SSE的服务端端点 var eventSource = new EventSource('/subscribe'); console.log("eventSource=", eventSource) // 监听message事件，接收从服务端发送的消息 eventSource.addEventListener('message', function(event) { var message = event.data; console.log("message=", message) var messageContainer = document.getElementById('message-container'); messageContainer.innerHTML += '<p>' + message + '</p>'; }); </script>

上述过程：客户端可以通过访问/subscribe接口来订阅SSE事件，服务器会返回一个SseEmitter对象。当有新消息到达时，调用SseEmitter对象的send()方法发送消息。

Web-Socket：

HTML代码： 

<script> // 创建WebSocket对象，并指定服务器的URL var socket = new WebSocket('ws://localhost:8080/上下文路径/channel/message/'); // 监听WebSocket的连接事件 socket.onopen = function(event) { console.log('WebSocket connected'); }; // 监听WebSocket的消息事件 socket.onmessage = function(event) { var message = event.data; var messageContainer = document.getElementById('message-container'); messageContainer.innerHTML += '<p>' + message + '</p>'; }; // 监听WebSocket的关闭事件 socket.onclose = function(event) { console.log('WebSocket closed'); }; // 发送消息到服务器 function sendMessage() { var messageInput = document.getElementById('message-input'); var message = messageInput.value; socket.send(message); messageInput.value = ''; } </script>

三、项目中使用的消息推送

例子1:Web-Socket

1.引入websocket依赖

<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-websocket</artifactId> </dependency> 

2.websocket配置

/ * @描述 开启WebSocket支持的配置类 * 自动注册使用@ServerEndpoint */ @Configuration public class WebSocketConfig { @Bean public ServerEndpointExporter serverEndpointExporter() { return new ServerEndpointExporter(); } } 

3.websocket服务器端代码

说明：@ ServerEndpoint 注解是一个类层次的注解，主要是将当前类定义成一个websocket服务器端, 注解的值将被用于监听用户连接的终端访问URL地址,客户端可以通过这个URL来连接到WebSocket服务器端

/ * 消息推送 / @ServerEndpoint("/channel/message/{user-id}") @Slf4j @Component @RequiredArgsConstructor public class TodoChannel implements ApplicationListener<FlowMessageEvent> { private static final Map<String, Set<Session>> SESSION_MAP = new ConcurrentHashMap<>(); private Session session; private String userId; @OnMessage public void onMessage(String message) { log.info("websocket消息(id={}): {}", this.session.getId(), message); } @OnOpen public void onOpen(Session session, @PathParam("user-id") String userId) { this.session = session; this.userId = userId; val sessionSet = SESSION_MAP.getOrDefault(this.userId, new CopyOnWriteArraySet<>()); sessionSet.add(session); SESSION_MAP.put(this.userId, sessionSet); log.info("websocket连接: id={}", this.session.getId()); val message = new MessageModel(); //往todoModel放业务数据 session.getAsyncRemote().sendText(JSON.toJSONString(message)); } @OnClose public void onClose(CloseReason closeReason) { val sessionSet = SESSION_MAP.get(this.userId); if (sessionSet != null) { sessionSet.remove(session); } log.info("websocket断开: id={} {}", this.session.getId(), closeReason); } @OnError public void onError(Throwable throwable) { log.warn("websocket异常: id={} throwable:", this.session.getId(), throwable); val sessionSet = SESSION_MAP.get(this.userId); if (sessionSet != null) { sessionSet.remove(this.session); } try { this.session.close(new CloseReason(CloseReason.CloseCodes.UNEXPECTED_CONDITION, throwable.getMessage())); } catch (IOException e) { log.error("websocket关闭失败", e); } } @Override public void onApplicationEvent(@NotNull FlowMessageEvent event) { Set<String> userIds = CastUtils.cast(event.getSource()); userIds.forEach(id -> { val sessionSet = SESSION_MAP.get(id); if (sessionSet == null) { return; } //业务处理todo sessionSet.forEach(s -> s.getAsyncRemote().sendObject(JSON.toJSON(message))); }); } @Scheduled(fixedRate = 24 * 60 * 60 * 1000L) public void sessionCleaner() { log.info("websocket message channel session清理"); val keyToClean = new HashSet<String>(); SESSION_MAP.forEach((k, v) -> { val sessionToClean = new HashSet<Session>(); v.forEach(s -> { if (!s.isOpen()) { sessionToClean.add(s); } }); v.removeAll(sessionToClean); if (v.isEmpty()) { keyToClean.add(k); } }); keyToClean.forEach(SESSION_MAP::remove); } @Data private static class MessageModel implements Serializable { private static final long serialVersionUID = 1L; private List<Info> list; private Integer size; @AllArgsConstructor @Value private static class Info implements Serializable { private static final long serialVersionUID = 1L; String type; String name; } } }

 4.事件类代码

/  * message事件  / public class FlowMessageEvent extends ApplicationEvent {     public FlowMessageEvent(Object source) {         super(source);     } } 

5.使用事件推送消息

applicationEventPublisher.publishEvent(new FlowMessageEvent(user));

例子2:RabbitMq：

1.引入rabbitmq依赖

2.编写rabbitmq配置类

/ * @author wux * @version 1.0.0 */ @SpringBootConfiguration @Slf4j public class RabbitMqConfig { @Value("${spring.rabbitmq.host:10.128.30.xxx}") private String host; @Value("${spring.rabbitmq.port:5672}") private int port; @Value("${spring.rabbitmq.username:guest}") private String username; @Value("${spring.rabbitmq.password:guest}") private String password; @Bean public ConnectionFactory connectionFactory() { CachingConnectionFactory factory = new CachingConnectionFactory(host, port); factory.setUsername(username); factory.setPassword(password); //连接工厂开启消息确认和消息返回机制 // factory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED); // factory.setPublisherReturns(true); return factory; } @Bean public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory) { SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory(); factory.setConnectionFactory(connectionFactory); //使用json 序列化和反序列化 factory.setMessageConverter(new Jackson2JsonMessageConverter()); //factory.setAcknowledgeMode(AcknowledgeMode.AUTO); return factory; } @Bean @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE) public RabbitTemplate rabbitTemplate() { RabbitTemplate template = new RabbitTemplate(); template.setConnectionFactory(connectionFactory()); template.setMessageConverter(new Jackson2JsonMessageConverter()); return template; } @Bean public Queue testDirectQueue() { return new Queue(RabbitMqConsts.TEST_QUE_PHM_WARN_INFO, true, false, false); } @Bean public DirectExchange testDirectExchange() { return new DirectExchange(RabbitMqConsts.TEST_EXC_PHM_WARN_INFO, true,false); } @Bean public Binding TestBinding() { return BindingBuilder.bind(testDirectQueue()).to(testDirectExchange()).with(RabbitMqConsts.TEST_KEY_PHM_WARN_INFO); } }

3.编写rabbitmq工具类

/ * @author wux * @version 1.0.0 * @description rabbit工具类 */ @Slf4j public class RabbitMqUtil { private static AmqpAdmin getAmqpAdmin() { return SpringContextUtils.getBean("amqpAdmin",AmqpAdmin.class); } / 通过amqpAdmin动态创建队列、交换机和绑定关系 ttlFlag :设置消息过期时间 */ public static void createQueueAndExchangeIfNeed(String businessName, ExchangeEnum typeEnum, Integer ttl) { String exchangeName = "exc_" + businessName; String queueName = "que_" + businessName; String routingKey = "key_" + businessName; if (CheckUtil.isNotEmpty(getQueueInfo(queueName))) { return; } //创建队列 Queue queue = createAndBindQueue(queueName, ttl); //创建交换机 Exchange exchange = createAndBindExchange(exchangeName, typeEnum); //绑定队列和交换机 switch (typeEnum){ case DIRECT: binding(queueName, exchangeName, routingKey, typeEnum); break; case FANOUT: fanoutBinding(queue, exchange); break; default: binding(queueName, exchangeName, routingKey, typeEnum); } } private static Queue createAndBindQueue(String queueName, Integer ttl) { Map<String, Object> arguments = new HashMap<>(); //设置过期时间，单位是毫秒 if (CheckUtil.isNotEmpty(ttl)) { arguments.put("x-message-ttl", ttl); } if (CheckUtil.isEmpty(queueName)) { log.error("队列名称为空！queueName=" + queueName); throw new BusinessException("队列名称为空！"); } Queue queue = new Queue(queueName, true, false, false, arguments); getAmqpAdmin().declareQueue(queue); return queue; } public static QueueInformation getQueueInfo (String queueName) { if (CheckUtil.isEmpty(queueName)) { return null; } return getAmqpAdmin().getQueueInfo(queueName); } private static Exchange createAndBindExchange(String exchangeName, ExchangeEnum typeEnum){ AbstractExchange exchange = null; switch (typeEnum){ case DIRECT: exchange = new DirectExchange(exchangeName, true, false); break; case TOPIC: exchange = new TopicExchange(exchangeName, true, false); break; case FANOUT: exchange = new FanoutExchange(exchangeName, true, false); break; case HEADERS: exchange = new HeadersExchange(exchangeName, true, false); break; default: exchange = new DirectExchange(exchangeName, true, false); } getAmqpAdmin().declareExchange(exchange); return exchange; } private static void binding(String queueName, String exchangeName, String routingKey, ExchangeEnum typeEnum) { //绑定队列和交换机 Binding binding = new Binding(queueName, Binding.DestinationType.QUEUE, exchangeName, routingKey, null); getAmqpAdmin().declareBinding(binding); } private static void fanoutBinding(Queue queue, Exchange exchange) { BindingBuilder.bind(queue).to(exchange); } }

4.监听和推送消息

/ * @author wux * @version 1.0.0 * @description 监听phm设备状态消息 */ @Component @Slf4j public class MotePhmDeviceStatesListener { private static final String CLASS_NAME = "MotePhmDeviceStatesListener"; @Autowired private Map<String, AssembleDeviceStatesStrategy> map = new ConcurrentHashMap<String, AssembleDeviceStatesStrategy>(); @Resource private MoteMessageService moteMessageService; @Autowired protected BeanMapper beanMapper; @RabbitHandler @RabbitListener(bindings = @QueueBinding(value=@Queue("que_phm_device_states"),exchange = @Exchange("exc_phm_device_states") ,key = "key_phm_device_states")) public void process(@Payload BaseMessage req, Message msg, Channel channel) { final String METHOD_NAME ="process"; log.info(CLASS_NAME + "-" + METHOD_NAME + "-start,req={},msg={}", req, msg); long deliverTag = msg.getMessageProperties().getDeliveryTag(); if (!MessageTypeEnum.DEVICE_STATES.getKey().equals(req.getType())) { log.warn(CLASS_NAME + "-" + METHOD_NAME + "-message=消息类型不匹配！"); //拒绝，重新回到队列 //channel.clearReturnListeners(); //channel.basicReject(deliverTag, true); return; } try { String service = MessageSubTypeEnum.getService(req.getSubType()); AssembleDeviceStatesStrategy strategy = map.get(service); BaseMessage reqMessage = beanMapper.map(req, BaseMessage.class); BaseMessage retMessage = strategy.assembleDeviceStates(reqMessage); strategy.sendMessage(retMessage, reqMessage, channel); } catch (Exception e) { log.error(CLASS_NAME + "-" + METHOD_NAME + "-异常, e={}", e); return; //拒绝，重新回到队列 //channel.basicNack(deliverTag, false,true); } } }

 public void sendMessage(BaseMessage retMessage, BaseMessage req, Channel channel) { retMessage.setDate(new Date()); retMessage.setDateStr(DateUtils.format(retMessage.getDate(), DateUtils.DATE_TIME_SECOND)); if (CheckUtil.isEmpty(req.getQueueName())) { log.warn("AssembleDeviceStatesStrategy" + "队列名称为空,req={}", req); return; } QueueInformation queueInfo = RabbitMqUtil.getQueueInfo(req.getQueueName()); if (CheckUtil.isEmpty(queueInfo) || CheckUtil.isEmpty(queueInfo.getName())) { log.warn("AssembleDeviceStatesStrategy" + "-" + "队列不存在!" + ",req={}", req); return; } // try { // long count = channel.messageCount(req.getQueueName()); // if (count >= 5000) { // channel.queueDelete(req.getQueueName()); // } // // } catch (IOException e) { // log.error("AssembleDeviceStatesStrategy" + "-" + "清除队列消息失败" + ",retMessage={}", retMessage, e); // } rabbitTemplate.convertAndSend(req.getQueueName(), retMessage); log.info("AssembleDeviceStatesStrategy" + "-" + "sendMessage推给前端信息" + ",retMessage={},req={}", retMessage, req); }

例子3:Kafka：

使用@KafkaListene(topics=”xxx”, groupId=”xxx”)  接受消息

四、消息中间件：RabbitMQ、RocketMQ、Kafka

高并发情况下，或者规模较大，推荐使用消息中间件，搭建一个公共平台，统一管理消息推送，项目层面进行隔离即可。

RabbitMQ可看：工作七年，对消息推送使用的一些经验和总结