SpringBoot整合RabbitMQ

参考文章：https://blog.csdn.net/qq_35387940/article/details/100514134

Direct Exchange
直连型交换机，根据消息携带的路由键将消息投递给对应队列。
大致流程，有一个队列绑定到一个直连交换机上，同时赋予一个路由键 routing key 。
然后当一个消息携带着路由值为X，这个消息通过生产者发送给交换机时，交换机就会根据这个路由值X去寻找绑定值也是X的队列。

Fanout Exchange
扇型交换机，这个交换机没有路由键概念，就算你绑了路由键也是无视的。 这个交换机在接收到消息后，会直接转发到绑定到它上面的所有队列。

Topic Exchange
主题交换机，这个交换机其实跟直连交换机流程差不多，但是它的特点就是在它的路由键和绑定键之间是有规则的。
简单地介绍下规则：

*  (星号) 用来表示一个单词 (必须出现的)
#  (井号) 用来表示任意数量（零个或多个）单词
通配的绑定键是跟队列进行绑定的，举个小例子
队列Q1 绑定键为 *.TT.*          队列Q2绑定键为  TT.#
如果一条消息携带的路由键为 A.TT.B，那么队列Q1将会收到；
如果一条消息携带的路由键为TT.AA.BB，那么队列Q2将会收到；

主题交换机是非常强大的，为啥这么膨胀？
当一个队列的绑定键为 "#"（井号） 的时候，这个队列将会无视消息的路由键，接收所有的消息。
当 * (星号) 和 # (井号) 这两个特殊字符都未在绑定键中出现的时候，此时主题交换机就拥有的直连交换机的行为。
所以主题交换机也就实现了扇形交换机的功能，和直连交换机的功能。

另外还有 Header Exchange 头交换机 ，Default Exchange 默认交换机，Dead Letter Exchange 死信交换机，这几个该篇暂不做讲述。

项目结构

image.png

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.1.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.wzf</groupId>
    <artifactId>websocket</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>websocket</name>
    <description>websocket</description>

    <properties>
        <java.version>1.8</java.version>
    </properties>

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

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

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

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

application.properties

spring.application.name=websocket-test
server.port=8888

spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123456

测试控制类RabbitController.java

package com.wzf.websocket.controller;


import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

@RestController
@RequestMapping("/rabbit")
public class RabbitController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/sendDirectMsg")
    public String sendDirectMsg(){
        Map<String, Object> msg = new HashMap<>();
        msg.put("id", UUID.randomUUID().toString());
        msg.put("data", "sendDirectMsg");
        rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting", msg);
        return "OK";
    }

    @GetMapping("/sendTopicMsg")
    public String sendTopicMsg(){
        Map<String, Object> msg = new HashMap<>();
        msg.put("id", UUID.randomUUID().toString());
        msg.put("data", "sendTopicMsg: man");
        rabbitTemplate.convertAndSend("TestTopicExchange", "topic.man", msg);
        return "OK";
    }

    @GetMapping("/sendTopicMsg2")
    public String sendTopicMsg2(){
        Map<String, Object> msg = new HashMap<>();
        msg.put("id", UUID.randomUUID().toString());
        msg.put("data", "sendTopicMsg: woman");
        rabbitTemplate.convertAndSend("TestTopicExchange", "topic.#", msg);
        return "OK";
    }

    @GetMapping("/sendFanoutMsg")
    public String sendFanoutMsg(){
        Map<String, Object> msg = new HashMap<>();
        msg.put("id", UUID.randomUUID().toString());
        msg.put("data", "sendFanoutMsg");
        rabbitTemplate.convertAndSend("TestFanoutExchange", null, msg);
        return "OK";
    }
}

直连型交换机

DirectRabbitConfig.java

package com.wzf.websocket.config;


import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 直连型交换机
 */
@Configuration
public class DirectRabbitConfig {

    // 队列起名：TestDirectQueue
    @Bean
    public Queue TestDirectQueue(){
        return new Queue("TestDirectQueue", true);
    }

    // 交换机起名：TestDirectExchange
    @Bean
    public DirectExchange TestDirectExchange(){
        return new DirectExchange("TestDirectExchange", true, false);
    }

    // 绑定：将队列和交换机绑定，并设置匹配键 TestDirectRouting
    @Bean
    public Binding  bindingDirect(){
        return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
    }

    @Bean
    public DirectExchange lonelyDirectExchange(){
        return new DirectExchange("lonelyDirectExchange");
    }

}

DirectReceiver.java

package com.wzf.websocket.component;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * 直连交换机是采用轮询的方式对消息进行消费，而且不存在重复消费
 */
@Component
@RabbitListener(queues = "TestDirectQueue")
public class DirectReceiver {

    @RabbitHandler
    public void process(Map msg) {
        System.out.println("DirectReceiver接收到的消息：" + msg.toString());
    }
}

DirectReceiver2.java

package com.wzf.websocket.component;


import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * 直连交换机是采用轮询的方式对消息进行消费，而且不存在重复消费
 */
@Component
@RabbitListener(queues = "TestDirectQueue")
public class DirectReceiver2 {

    @RabbitHandler
    public void process(Map msg) {
        System.out.println("DirectReceiver2接收到的消息：" + msg.toString());
    }
}

效果

image.png

主题交换机

TopicRabbitConfig.java

package com.wzf.websocket.config;


import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 主题型交换机
 */
@Configuration
public class TopicRabbitConfig {


    @Bean
    public Queue manTopicQueue(){
        return new Queue("topic.man", true);
    }

    @Bean
    public Queue womanTopicQueue(){
        return new Queue("topic.woman", true);
    }


    @Bean
    public TopicExchange TestTopicExchange(){
        return new TopicExchange("TestTopicExchange", true, false);
    }


    @Bean
    public Binding  bindingMan(){
        return BindingBuilder.bind(manTopicQueue()).to(TestTopicExchange()).with("topic.man");
    }

    @Bean
    public Binding  bindingTotal(){
        return BindingBuilder.bind(womanTopicQueue()).to(TestTopicExchange()).with("topic.#");
    }
}

TopicManReceiver.java

package com.wzf.websocket.component;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * 主题交换机
 */
@Component
@RabbitListener(queues = "topic.man")
public class TopicManReceiver {

    @RabbitHandler
    public void process(Map msg) {
        System.out.println("TopicManReceiver接收到的消息：" + msg.toString());
    }

    @RabbitHandler
    public void process2(String msg) {
        System.out.println("TopicManReceiver接收到的消息：" + msg);
    }
}

TopicTotalReceiver.java

package com.wzf.websocket.component;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * 主题交换机
 */
@Component
@RabbitListener(queues = "topic.woman")
public class TopicTotalReceiver {

    @RabbitHandler
    public void process(Map msg) {
        System.out.println("TopicTotalReceiver接收到的消息：" + msg.toString());
    }

    @RabbitHandler
    public void process2(String msg) {
        System.out.println("TopicTotalReceiver接收到的消息：" + msg);
    }
}

效果

image.png

扇形交换机

FanoutRabbitConfig.java

package com.wzf.websocket.config;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 扇形交换机
 */
@Configuration
public class FanoutRabbitConfig {
    @Bean
    public Queue aFanoutQueue() {
        return new Queue("fanout.A", true);
    }

    @Bean
    public Queue bFanoutQueue() {
        return new Queue("fanout.B", true);
    }

    @Bean
    public Queue websocketFanoutQueue() {
        return new Queue("fanout.websocket", true);
    }

    @Bean
    public FanoutExchange TestFanoutExchange() {
        return new FanoutExchange("TestFanoutExchange", true, false);
    }

    @Bean
    public Binding bindingFanoutA() {
        return BindingBuilder.bind(aFanoutQueue()).to(TestFanoutExchange());
    }

    @Bean
    public Binding bindingFanoutB() {
        return BindingBuilder.bind(bFanoutQueue()).to(TestFanoutExchange());
    }

    @Bean
    public Binding bindingFanoutWebsocket() {
        return BindingBuilder.bind(websocketFanoutQueue()).to(TestFanoutExchange());
    }
}

FanoutReceiverA.java

package com.wzf.websocket.component;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * 扇形交换机
 */
@Component
@RabbitListener(queues = "fanout.A")
public class FanoutReceiverA {

    @RabbitHandler
    public void process(Map msg) {
        System.out.println("FanoutReceiverA接收到的消息：" + msg.toString());
    }

    @RabbitHandler
    public void process2(String msg) {
        System.out.println("FanoutReceiverA接收到的消息：" + msg);
    }
}

FanoutReceiverB.java

package com.wzf.websocket.component;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * 扇形交换机
 */
@Component
@RabbitListener(queues = "fanout.B")
public class FanoutReceiverB {

    @RabbitHandler
    public void process(Map msg) {
        System.out.println("FanoutReceiverB接收到的消息：" + msg.toString());
    }

    @RabbitHandler
    public void process2(String msg) {
        System.out.println("FanoutReceiverB接收到的消息：" + msg);
    }
}

效果

image.png

websocket

WebSocketConfig.java

package com.wzf.websocket.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;

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

WebSocketServer.java

package com.wzf.websocket.component;

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.concurrent.CopyOnWriteArraySet;

@Slf4j
@Component
@ServerEndpoint("/websocket/{sid}")
public class WebSocketServer {

    //静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
    private static int onlineCount = 0;

    //concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
    private static CopyOnWriteArraySet<WebSocketServer> webSocketSet = new CopyOnWriteArraySet<WebSocketServer>();

    //与某个客户端的连接会话，需要通过它来给客户端发送数据
    private Session session;

    //接收sid
    private String sid="";

    /**
     * 连接建立成功调用的方法*/
    @OnOpen
    public void onOpen(Session session,@PathParam("sid") String sid) throws InterruptedException {
        this.session = session;
        webSocketSet.add(this);     //加入set中
        addOnlineCount();           //在线数加1
        log.info("有新窗口开始监听:"+sid+",当前在线人数为" + getOnlineCount());
        this.sid=sid;
        try {
            //模拟服务器向客户端发消息
            sendMessage("服务器：恭喜你，连接成功");
        } catch (IOException e) {
            log.error("websocket IO异常");
        }
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        webSocketSet.remove(this);  //从set中删除
        subOnlineCount();           //在线数减1
        log.info("有一连接关闭！当前在线人数为" + getOnlineCount());
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息*/
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("收到来自窗口"+sid+"的信息:"+message);
//        for (WebSocketServer item : webSocketSet) {
//            try {
//                item.sendMessage(message);
//            } catch (IOException e) {
//                e.printStackTrace();
//            }
//        }
    }

    /**
     *连接发生错误时
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("发生错误");
        error.printStackTrace();
    }

    /**
     * 实现服务器主动推送
     */
    public void sendMessage(String message) throws IOException {
        if(this.session != null){
            this.session.getBasicRemote().sendText(message);
        }
    }


    /**
     * 实现服务器主动向某一客户端推送消息
     * */
    public void sendInfo(String message,@PathParam("sid") String sid) throws IOException {
        log.info("推送消息到窗口"+sid+"，推送内容:"+message);
        for (WebSocketServer item : webSocketSet) {
            try {
                if(sid == null){
                    item.sendMessage(message);
                }else if(item.sid.equals(sid)){
                    item.sendMessage(message);
                }
            } catch (IOException e) {
                continue;
            }
        }
    }

    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

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

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

WebsocketReceiver.java

package com.wzf.websocket.component;


import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Map;

/**
 * websocket消费消息对垒的消息
 */
@Component
@RabbitListener(queues = "fanout.websocket")
public class WebsocketReceiver {

    @Autowired
    private WebSocketServer webSocketServer;

    @RabbitHandler
    public void process(Map msg) throws IOException {
        if(msg.get("sid") != null){
            webSocketServer.sendInfo(msg.toString(), msg.get("sid").toString());
        }else{
            webSocketServer.sendInfo(msg.toString(), null);
        }
    }

    @RabbitHandler
    public void process2(String msg) throws IOException {
        webSocketServer.sendInfo(msg, null);
    }
}

前端

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta http-equiv="X-UA-Compatible" content="IE=edge">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>websocket</title>
</head>
<body>
    <script src="./Stomp.js"></script>
        <script>
            // 通过java websocket消费消息
            // let url = "ws://127.0.0.1:8888/websocket/1";
            // let socket = new WebSocket(url);
            // socket.onopen = function (){
            //  console.log("连接成功！");
            //  socket.send("客户端发送的消息");
            // };
            // socket.onmessage = function (msg){
            //  console.log(msg.data);
            // };
            // socket.onerror = function (error){
            //  console.log(error);
            // };

            // 直接消费rabbitmq的消息
            let url = "ws://127.0.0.1:15674/ws";
            let client = Stomp.client(url);
            client.connect("admin","123456",function(){
                console.log("连接成功！");
                // 直接监听某个队列消息：/amq/queue/queue名称
                let sub = client.subscribe("/amq/queue/TestDirectQueue",function(message){
                    console.log("TestDirectQueue接收到消息：", message);
                })
                // 直连交换机: /exchange/exchange名称/routingKey
                let sub1 = client.subscribe("/exchange/TestDirectExchange/TestDirectRouting",function(message){
                    console.log("TestDirectExchange接收到消息：", message);
                })
                // 主题交换机: /exchange/exchange名称/routingKey
                let sub2 = client.subscribe("/exchange/TestTopicExchange/topic.man",function(message){
                    console.log("TestTopicExchange接收到消息：", message);
                })
                // 扇形交换机: /exchange/exchange名称/routingKey
                let sub3 = client.subscribe("/exchange/TestFanoutExchange",function(message){
                    console.log("TestFanoutExchange接收到消息：", message);
                })
            }, function(error){
                console.log(error);
            });
        </script>
</body>
</html>

效果

image.png
image.png