Rabbitmq+WebSocket实现消息推送给web前端

心已赠人 2021-09-15 10:04 493阅读 0赞

后台应该框架使用的是springboot,必须的依赖如下：

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-websocket</artifactId>
    </dependency>
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>4.1.0</version>
    </dependency>

添加config文件：

//@Configuration public class WebSocketConfig {
        @Bean
        public ServerEndpointExporter serverEndpointExporter() {
            return new ServerEndpointExporter();
        }
    }

添加wobsocket相关文件类：

//该注解用来指定一个URI，客户端可以通过这个URI来连接到WebSocket。类似Servlet的注解mapping。无需在web.xml中配置。 @ServerEndpoint(value = "/websocket")
    @Component
    public class MyWebSocket {
        @Value("${spring.rabbitmq.host:172.16.1.29}")
        private String rabbitHost;
        @Value("${spring.rabbitmq.port:5672}")
        private Integer rabbitPort;
        @Value("${spring.rabbitmq.username:admin}")
        private String rabbitUser;
        @Value("${spring.rabbitmq.password:a123456}")
        private String rabbitPass;
    
        private static Logger log = LogManager.getLogger(MyWebSocket.class);
        //静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。 // @Autowired // public TerminalService terminalServiceInWebSocket; //与某个客户端的连接会话，需要通过它来给客户端发送数据 private Session session;
        private static CopyOnWriteArraySet<Session> sessions = new CopyOnWriteArraySet<Session>();
    
        private static int onlineCount = 0;
    
        //concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。若要实现服务端与单一客户端通信的话，可以使用Map来存放，其中Key可以为用户标识 private static CopyOnWriteArraySet<MyWebSocket> webSocketSet = new CopyOnWriteArraySet<MyWebSocket>();
    
    
    
        /** * 连接建立成功调用的方法 * * @param session 可选的参数。session为与某个客户端的连接会话，需要通过它来给客户端发送数据 */ @OnOpen
        public void onOpen(Session session) {
            this.session = session;
            sessions.add(session);
            webSocketSet.add(this);     //加入set中 addOnlineCount();           //在线数加1 log.info("有新连接加入！当前在线人数为" + getOnlineCount());
            String QUEUE_NAME = "closedPlace";
            try {
                //创建连接连接到MabbitMQ ConnectionFactory factory = new ConnectionFactory();
                //设置MabbitMQ所在主机ip或者主机名 factory.setHost("172.16.1.29");
    // factory.setPort(rabbitPort); factory.setUsername("admin");
                factory.setPassword("a123456");
                //创建一个连接 Connection connection = factory.newConnection();
                //创建一个频道 Channel channel = connection.createChannel();
                //指定一个队列 /* durable：true、false.true：在服务器重启时，能够存活 exclusive ：是否为当前连接的专用队列，在连接断开后，会自动删除该队列，生产环境中应该很少用到吧。 autodelete：当没有任何消费者使用时，自动删除该队列。this means that the queue will be deleted when there are no more processes consuming messages from it. */ channel.queueDeclare(QUEUE_NAME, true, false, false, null);
                //每次从队列获取的数量,保证一次只分发一个 channel.basicQos(1);
                log.info(" [*] Waiting for messages.");
                //监听队列 Consumer consumer = new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                        //处理监听得到的消息 String message = null;
                        try {
                            message = new String(body, "UTF-8");
                            //消息处理逻辑 sendMessage(message);
                        } catch (UnsupportedEncodingException e) {
                            e.printStackTrace();
                            channel.abort();
                        } finally {
                            log.info("[x] Done.");
                            channel.basicAck(envelope.getDeliveryTag(), false);
                        }
                        log.info("[x] Received '" + message + "'");
                    }
                };
    
                //autoAck是否自动回复，如果为true的话，每次生产者只要发送信息就会从内存中删除，那么如果消费者程序异常退出， // 那么就无法获取数据，我们当然是不希望出现这样的情况，所以才去手动回复，每当消费者收到并处理信息然后在通知 // 生成者。最后从队列中删除这条信息。如果消费者异常退出，如果还有其他消费者，那么就会把队列中的消息发送给其 // 他消费者，如果没有，等消费者启动时候再次发送。 boolean autoAck = false;
                //消息消费完成确认 channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
    
        }
    
        /** * 连接关闭调用的方法 */ @OnClose
        public void onClose() {
            webSocketSet.remove(this);  //从set中删除 sessions.remove(session);
            subOnlineCount();           //在线数减1 log.info("有一连接关闭！当前在线人数为" + getOnlineCount());
        }
    
        /** * 收到客户端消息后调用的方法 * * @param message 客户端发送过来的消息 * @param session 可选的参数 */ @OnMessage
        public void onMessage(String message, Session session) {
            log.info("来自客户端的消息:" + message);
            //群发消息 for (MyWebSocket item : webSocketSet) {
                try {
                    item.sendMessage(message);
                } catch (IOException e) {
                    e.printStackTrace();
                    continue;
                }
            }
        }
    
        /** * 发生错误时调用 * * @param session * @param error */ @OnError
        public void onError(Session session, Throwable error) {
            log.error("发生错误");
            error.printStackTrace();
        }
    
        /** * 这个方法与上面几个方法不一样。没有用注解，是根据自己需要添加的方法。 * * @param message * @throws IOException */ public void sendMessage(String message) throws IOException {
            //阻塞式的(同步的) if (sessions.size() != 0) {
                for (Session s : sessions) {
                    if (s != null) {
                        s.getBasicRemote().sendText(message);
                    }
                }
            }
    
            //非阻塞式的（异步的） // this.session.getAsyncRemote().sendText(message); log.info("[x] 推送消息"+message);
        }
    
        public static synchronized int getOnlineCount() {
            return onlineCount;
        }
    
        public static synchronized void addOnlineCount() {
            MyWebSocket.onlineCount++;
        }
    
        public static synchronized void subOnlineCount() {
            MyWebSocket.onlineCount--;
        }
    }

添加订阅配置：

@Configuration
    public class RabbitTopic {
        final static String message = "closedPlace";
        final static String messages = "topic.messages";
    
        //创建队列 @Bean
        public Queue queueMessage() {
            return new Queue(RabbitTopic.message);
        }
    
        //创建队列 @Bean
        public Queue queueMessages() {
            return new Queue(RabbitTopic.messages);
        }
    
        //创建交换器 @Bean
        TopicExchange exchange() {
            return new TopicExchange("topicExchange");
        }
    
        //对列绑定并关联到ROUTINGKEY @Bean
        Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
            return BindingBuilder.bind(queueMessage).to(exchange).with(RabbitTopic.message);
        }
    
        //对列绑定并关联到ROUTINGKEY @Bean
        Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
            return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");//*表示一个词,#表示零个或多个词 }

web的html文件：

<!DOCTYPE html>
    <html>  
    <head>  
    	 <meta charset="UTF-8">
        <title>WebSocket</title>  
    </head>  
    <body>  
    <div id="main">  
        <div id="message"></div>  
    </div>  
    </body>  
    <script type="text/javascript">  
        var websocket = null;  
      
        //判断当前浏览器是否支持WebSocket  
        if ('WebSocket' in window) {  
            websocket = new WebSocket("ws://192.168.2.59:8080/springboot/websocket");  
        }  
        else {  
            alert('当前浏览器 不支持WebSocket')  
        }  
      
        //连接发生错误的回调方法  
        websocket.onerror = function () {  
            setMessageInnerHTML("连接发生错误");  
        };  
      
        //连接成功建立的回调方法  
        websocket.onopen = function () {  
            setMessageInnerHTML("连接成功");  
        }  
      
        //接收到消息的回调方法，此处添加处理接收消息方法，当前是将接收到的信息显示在网页上  
        websocket.onmessage = function (event) {  
            setMessageInnerHTML(event.data);  
        }  
      
        //连接关闭的回调方法  
        websocket.onclose = function () {  
            setMessageInnerHTML("连接关闭,如需登录请刷新页面。");  
        }  
      
        //监听窗口关闭事件，当窗口关闭时，主动去关闭websocket连接，防止连接还没断开就关闭窗口，server端会抛异常。  
        window.onbeforeunload = function () {  
            closeWebSocket();  
        }  
      
        //将消息显示在网页上，如果不需要显示在网页上，则不调用该方法  
        function setMessageInnerHTML(innerHTML) {  
            document.getElementById('message').innerHTML += innerHTML + '<br/>';  
        }  
    </script>  
    </html>