rabbitmq+websocket(SpringBoot版)實現分布式消息推送

本來想用websocket做一個消息推送 可是分布式環境下不支持session共享因為服務器不同

所以采用 rabbitMQ+webSocket實現分布式消息推送

生產者將消息 發送給 rabbitMQ 的 virtual-host:/(頂極路由) 再由它路由到交換機 最終由交換機通過路由鍵指定具體的管道

消費者監聽指定的管道獲取消息

最終將獲取的消息 交給 webSocket 被@OnMessage注解標識的方法

每次消費一條消息交給　被@OnMessage注解標識的方法 返回給前台 

實現分布式實時推送

1.配置rabbitMQ

消息生產者

1.1pom.xml

<!--引入rabbitmq依賴-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

server:
  port: 5002

spring:
  rabbitmq:
    host: localhost
    #賬號密碼 默認有的
    username: guest
    password: guest
    #rbbitmq虛擬主機路徑
    virtual-host: /
    #rabbitmq的端口號 也是默認的
    port: 5672

@SpringBootApplication
@MapperScan(basePackages = "com.supplychain.dao")
@EnableRabbit/**開啟rabbitmq*/
public class ThumbsupServer5002_App {

    public static void main(String[]args){

        SpringApplication.run(ThumbsupServer5002_App.class,args);

    }

    /**消息的轉換器
     * 設置成json 並放入到Spring中
     * */
    @Bean
    public MessageConverter messageConverter(){

        return new Jackson2JsonMessageConverter();

    }
}

測試發送消息

@RunWith(SpringRunner.class)
@SpringBootTest
public class ThumbsupServer5002_AppTest {


    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void contextLoads() {

        UserTest userTest = new UserTest("hao", "[e-mail]");

        /**1.指定發送的交換機
         *      發送的消息會先發送給 virtual-host: /(頂級路由) 再由它到交換機
         *      由交換機通過路由鍵指定給具體的管道
         *
         * 2.路由鍵
         *      有的交換機需要路由鍵 有的不需要(發送給交換機的消息會被發送給所有管道)
         *
         * 3.發送的消息
         *      如果是對象的話必須實現序列化接口因為網絡傳輸只能傳二進制
         *
         * */
        rabbitTemplate.convertAndSend("userTest-exchange", "userTest-key", userTest);
    }

}

2.消息消費者

同樣是pom.xml需要引入rabbitMQ依賴

<!--引入rabbitmq依賴-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

同樣需要配置application.yml

spring:
  rabbitmq:
    host: 127.0.0.1
    #賬號密碼 默認有的
    username: guest
    password: guest
    #rbbitmq虛擬主機路徑
    virtual-host: /
    #rabbitmq的端口號 也是默認的
    port: 5672
    listener:
      simple:
        acknowledge-mode: manual #手動接受數據
        #max-concurrency: 10 #最大並發
        #prefetch: 1 #限流

同樣主啟動類中需要開啟RabbitMQ

@SpringBootApplication
@EnableRabbit
public class MessageServer5003_App {

    public static void main(String[]args){

        SpringApplication.run(MessageServer5003_App.class,args);

    }

    /**這里也需要設置消息轉換類型
     * 和發送的消息類型一定要對應
     * 不然對象接受json啟動主程序類時就會報錯
     * */
    @Bean
    public MessageConverter messageConverter(){

        return new Jackson2JsonMessageConverter();

    }

}

下面到了整合的環節了

@ServerEndpoint(value = "/websocket")
@Component
public class WebSocketServer {

    //靜態變量 用於記錄當前在線連接數 應該把它設計成線程安全
    private static int onlineCount=0;

    /**Concurrent包下的 寫時復制Set 用它作於存儲客戶端對應的MyWebSocket對象*/
    private static CopyOnWriteArraySet<WebSocketServer> webSocketSet= new CopyOnWriteArraySet<WebSocketServer>();


    /**與某個客戶端的鏈接會話,需要通過它來給客戶端發送數據*/

    private Session session;
    /**
     * 參數1:Message 可以獲得消息的內容字節 還可以獲得消息的其他屬性
     * 參數2:可以寫確定接受的參數類型比如User
     * 參數3:Channel 通道
     *      com.rabbitmq.client.Channel必須是這個包下
     *      通過這個參數可以拒絕消息
     *      讓rabbitmq再發給別的消費者
     *
     * 使用@RabbitListener 可以綁定交換機  路由鍵 管道
     *
     */
    @RabbitListener(bindings = @QueueBinding(
         value = @Queue(value = "userTest-queue",durable = "true"),
         exchange = @Exchange(name = "userTest-exchange",durable = "true",type = "direct"),
         key = "userTest-key"
        )
    )
    @RabbitHandler//注解意思:如果有消息過來 需要消費的時候才會調用該方法
    /**如果已知傳遞的參數是 UserTest對象可以通過該注解
     * 消息頭需要用map接受
     * 既然是手動接受消息 就需要設置channel
     * */
    public void receiveUserMessage(@Payload UserTest userTest, @Headers Map<String,Object> headers, Channel channel) throws IOException {
        //sendMessage(message.toString());
        System.out.println("UserTest對象"+userTest);
        onMessage(userTest.toString());//調用消息方法將數據船體給他

        Long deliveryTag= (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
        //手動接受並告訴rabbitmq消息已經接受了  deliverTag記錄接受消息 false不批量接受
        channel.basicAck(deliveryTag,false);

        /**
         * basicReject()
         * 參數1: 消息標簽
         * 參數2: true 將消息從新放入隊列  false 接受到並將消息拋棄
         *
         *
        try {
            channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
            System.out.println(message);
        } catch (IOException e) {
            e.printStackTrace();
        }
     */

    }

    /**服務器端推送消息*/
    public void sendMessage(String message){
        try {
            System.out.println("session可否顯示出來"+session);
            this.session.getBasicRemote().sendText(message);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 連接建立成功調用的方法
     * */
    @OnOpen
    public void onOpen(Session session){
        this.session=session;
        webSocketSet.add(this);
        System.out.println("有新的連接加入!當前在線人數為"+getOnlineCount());
        System.out.println(session);
    }

    /**
     * 連接關閉調用的方法
     * */
    @OnClose
    public void onClose(){
        /**從安全Set中 移除當前連接對象*/
        webSocketSet.remove(this);
        subOnlineCount();
        System.out.println("有一連接關閉!當前在線人數為"+getOnlineCount());
    }



    @OnMessage
    public void onMessage(String message){

        System.out.println("來自客戶端的消息:"+message);

        for (WebSocketServer webSocketServer:webSocketSet){
            webSocketServer.sendMessage(message);
        }

    }


    public static int getOnlineCount() {
        return onlineCount;
    }

    public static synchronized void addOnlineCount() {
        WebSocketServer.onlineCount++;
    }

    public static synchronized void subOnlineCount() {
        WebSocketServer.onlineCount--;
    }



}

websocket前端

websocket是html5提出的協議屬於雙工通信 前端發送一次請求告訴服務器需要將http協議升級成tcp長連接

后面服務端直接給前端推送消息就可以了 從以前的一次請求一次響應 服務端被動式 變成 一次請求服務端可以無限響應

<script>
        var socket;
        console.log(typeof socket)
        if (typeof(WebSocket)=="undefined"){
            alert("您的瀏覽器不支持WebSocket");
        }else{
            alert("您的瀏覽器支持WebSocket");

           socket=new WebSocket("ws://localhost:5003/websocket");

           socket.onopen=function () {
               console.log("Socket 已打開");
           };

            //獲得消息事件
            socket.onmessage = function(msg) {
                console.log(msg.data);
                //發現消息進入    調后台獲取
                //getCallingList();
            };

            //關閉事件
            socket.onclose = function() {
                console.log("Socket已關閉");
            };
            //發生了錯誤事件
            socket.onerror = function() {
                alert("Socket發生了錯誤");
            };
            /**
            $(window).unload(function(){
                socket.close();
            });
             */
        }
    </script>