websocket stomp与RabbitMQ结合

妖狐艹你老母 2023-06-30 10:57 4阅读 0赞

**一、配置application.properties**

#rabbitmq
    spring.rabbitmq.host=127.0.0.1
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest

**二、stomp的config**

package com.xxxx.checkandbigdataquery.stompmq;
    
    
    import org.springframework.context.annotation.Configuration;
    import org.springframework.messaging.simp.config.MessageBrokerRegistry;
    import org.springframework.web.socket.config.annotation.AbstractWebSocketMessageBrokerConfigurer;
    import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
    import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
    import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;
    
    @Configuration
    @EnableWebSocketMessageBroker
    public class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer implements WebSocketMessageBrokerConfigurer { 
    
    
        @Override
        public void configureMessageBroker(MessageBrokerRegistry registry) { 
            /** * 订阅来自"/topic"和"/user"的消息， * /topic 单聊 * /all 群聊 * 收消息 */
            registry.enableSimpleBroker("/topic","/all");
            /** * 客户端发送过来的消息，需要以"/app"为前缀，再经过Broker转发给响应的Controller */
            registry.setApplicationDestinationPrefixes("/app");
            /** * 配置消息代理 * 使用RabbitMQ做为消息代理，替换默认的Simple Broker * 设置目的地前缀 */
            registry
                    .enableStompBrokerRelay("/exchange","/topic","/queue","/amq/queue")
                    .setRelayHost("192.168.0.113")
                    .setClientLogin("test")
                    .setClientPasscode("test")
                    .setSystemLogin("test")
                    .setSystemPasscode("test")
                    .setSystemHeartbeatSendInterval(5000)
                    .setSystemHeartbeatReceiveInterval(4000);
    
            // 第一行，启动STOMP中继代理
            registry.enableStompBrokerRelay("/topic", "/queue")
                    .setRelayHost("localhost")
                    .setRelayPort(61613)
                    .setClientLogin("guest")
                    .setClientPasscode("guest");
        }
    
        @Override
        public void registerStompEndpoints(StompEndpointRegistry registry) { 
            /** * 路径"/websocket"被注册为STOMP端点，对外暴露，客户端通过该路径接入WebSocket服务 */
            registry.addEndpoint("/websocket").setAllowedOrigins("*").withSockJS();
            registry.addEndpoint("/websocket-rabbitmq").setAllowedOrigins("*").withSockJS();
        }
    
    
    }

**三、rabbitmq的配置**

package com.xxxx.checkandbigdataquery.stompmq;
    
    import org.springframework.amqp.core.Binding;
    import org.springframework.amqp.core.BindingBuilder;
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.core.TopicExchange;
    import org.springframework.context.annotation.Bean;
    
    public class RabbitConfigure { 
    
    
        // 队列名称
        public final static String SPRING_BOOT_QUEUE = "spring-boot-queue";
        // 交换机名称
        public final static String SPRING_BOOT_EXCHANGE = "spring-boot-exchange";
        // 绑定的值
        public static final String SPRING_BOOT_BIND_KEY = "spring-boot-bind-key";
    
    
    
        /** * 定义队列： * @return */
        @Bean
        Queue queue() { 
            return new Queue(SPRING_BOOT_QUEUE, false);
        }
    
        /** * 定义交换机 * @return */
        @Bean
        TopicExchange exchange() { 
            return new TopicExchange(SPRING_BOOT_EXCHANGE);
        }
    
        /** * 定义绑定 * @param queue * @param exchange * @return */
        @Bean
        Binding binding(Queue queue, TopicExchange exchange) { 
            return BindingBuilder.bind(queue).to(exchange).with(SPRING_BOOT_BIND_KEY);
        }
    
    }

**四、rabbitmq订阅消息**

package com.xxxx.checkandbigdataquery.stompmq;
    
    import org.springframework.amqp.core.Queue;
    import org.springframework.amqp.rabbit.annotation.RabbitHandler;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.context.annotation.Bean;
    import org.springframework.stereotype.Component;
    
    
    @Component
    //@RabbitListener(queues = RabbitConfigure.SPRING_BOOT_QUEUE)
    public class ReceiveMsg { 
    
    
        /** * 获取信息: * queue也可以支持RabbitMQ中对队列的模糊匹配 * @param content */
        @RabbitListener(queues = RabbitConfigure.SPRING_BOOT_QUEUE)
        public void receive(String content) { 
            System.out.println(content);
        }
    
        /** * //@RabbitListener(queues = RabbitConfigure.SPRING_BOOT_QUEUE) 放在类的头上 * 获取信息: * queue也可以支持RabbitMQ中对队列的模糊匹配 * @param msgContent */
        @RabbitHandler
        public void receiveMsgContent(String msgContent) { 
            System.out.println(msgContent);
        }
    
        /** * 转发 * @return */
        @Bean(name = RabbitConfigure.SPRING_BOOT_QUEUE)
        public Queue queueMessage() { 
            return new Queue("topic.message");
        }
    
    }

**五、controller测试**

package com.xxxx.checkandbigdataquery.stompmq;
    
    
    import org.springframework.amqp.core.AmqpAdmin;
    import org.springframework.amqp.core.AmqpTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.messaging.simp.SimpMessagingTemplate;
    import org.springframework.scheduling.annotation.EnableScheduling;
    import org.springframework.scheduling.annotation.Scheduled;
    import org.springframework.stereotype.Component;
    import org.springframework.stereotype.Controller;
    import java.text.DateFormat;
    import java.text.SimpleDateFormat;
    import java.util.Date;
    
    
    /** * @program: znkf * @description: * @author: Mr.Xiong * @create: 2019-07-10 14:50 **/
    @Controller
    @EnableScheduling //开启spring定时任务
    @Component  //可加可不加
    public class ClientWebSocketController { 
    
        @Autowired
        private SimpMessagingTemplate messagingTemplate;
    
    
        /** * 此接口默认实现只有一个，且是RabbitAdmin，通过源码发现其内部实现实际是RabbitTemplate。所以AmqpAdmin和AmqpTemplate当前两者本质是相同的 */
        @Autowired
        private AmqpAdmin amqpAdmin;
    
        /** * 此接口的默认实现是RabbitTemplate，目前只有一个实现 */
        @Autowired
        private AmqpTemplate amqpTemplate;
    
    
        /** * 测试定时推送消息 * 先与前端进行联调 */
        @Scheduled(cron = "1/4 * * * * *")
        public void callback() { 
            // 发现消息
            DateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            messagingTemplate.convertAndSend("/all", "定时推送消息时间: " + df.format(new Date()));
        }
    
        /** * 通过rabbitmq进行推送消息 */
        @Scheduled(cron = "1/4 * * * * *")
        public void sendTest() { 
            DateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            amqpTemplate.convertAndSend(RabbitConfigure.SPRING_BOOT_EXCHANGE, RabbitConfigure.SPRING_BOOT_BIND_KEY,df.format(new Date()));
        }
    
    }