• RabbitMQ（四） RabbitMQ的七种消息收发方式和四种交换机

    文章目录

    注意：消费者并发量的控制

    关于消费者并发量的控制：concurrency 属性

      /**
     * concurrency = "10"，这个表示有十个消费者，十个并发消费
     * @param msg
     */
    @RabbitListener(queues = "hello-queue",concurrency = "10")
       public  void  handleMsg(String msg){
           logger.info("msg:{}",msg);
       }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    连接依然是一个
    在这里插入图片描述
    但是channels对应的连接却是十一个：
    10+1
    在这里插入图片描述

    一. 七种消息收发方式

    1.1. Hello World

    上一篇文章的例子
    在这里插入图片描述

    1.先由生产者发送到消息交换机（有默认，可配置）
    2.消息交换机根据既定的策略帮你把消息路由到不同的队列上面去(意味着消息收发的形式更加的多样)
    3.消息消费者就去监听不同的队列

    1.2.work

    默认情况下，一个消费者中，开启一个线程进行消息的处理，意味着同一时间只能处理一条消息。

    我们可以通过配置，让一个消费者中，同时存在多个子线程，每一个线程都可以去消费消息。

    在这里插入图片描述
    即消费者类中，在添加一个消费者即可
    关键代码 RabbitConsumer类中

    一个生产者，一个默认的交换机，一个队列，两个消费者
    即两个消费者监听同一个队列

    @Component
public class RabbitConsumer {
       private static  final Logger logger = LoggerFactory.getLogger(RabbitConsumer.class);

    /**
     * @RabbitListener(queues = "hello-queue")
     * 通过这个注解，来指定该方法需要监听的消息队列，注解中的参数就是要去监听的消息队列的名称。
     * @param msg
     */
    @RabbitListener(queues = "hello-queue")
       public  void  handleMsg(String msg){
           logger.info("msg1:{}",msg);
       }


    @RabbitListener(queues = "hello-queue")
    public  void  handleMsg2(String msg){
        logger.info("msg:{}",msg);
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20

    界面显示的是两个通道
    在这里插入图片描述

    1.3 Publish/Subscribe

    一个生产者，一个交换机，两个队列，两个消费者

    在这里插入图片描述

    1.声明一个Fanout类型的exchange，并且将exchange和queue绑定在一起，绑定的方式就是直接绑定。
    2.让生产者创建一个exchange并且指定类型，和一个或多个队列绑定到一起。

    Publish/Subscribe的四种交换机

    在这里插入图片描述

    1.3.1 Direct直连交换机

    DirectExchange 的路由策略是将消息队列绑定到一个DirectExchange 上，当一条消息到达DirectExchange 时会被转发到与该条消息routing key相同的Queue上。

    例如消息队列名为“hello-queue”，则routingkey为“hello-queue”的消息会被该消息队列接收。

    具体步骤
    1.发送的时候给消息设置一个东西叫做routingkey
    2.生产者把消息发到交换机上面的时候，交换机会找一个和routingkey同名的队列把这条消息给他转发出去

    DirectConfig类

    步骤:

    1. 定义列 和交换机名字
    2. 列和交换机注入bean容器
    3. 绑定方法注入bean容器
    @Configuration
public class DirectConfig {
    //定义两个队列
    public  static final String MY_QUEUE_NAME_02 ="my queue name 02";
    public  static final String MY_QUEUE_NAME_03 ="my queue name 03";
    //交换机的名字
    public  static final String DIRECT_EXCHANGE_NAME="direct exchange name";


    @Bean
    Queue queue02(){
        return new Queue(MY_QUEUE_NAME_02,true,false,false);
    }

    @Bean
    Queue queue03(){
        return  new Queue(MY_QUEUE_NAME_03,true,false,false);
    }

    @Bean
    DirectExchange directExchange(){
        /**
         * 1.交换机的名字
         * 2. 交换机是否持久化
         * 3. 没有队列的时候，交换机是否自动化删除
         */
        return  new DirectExchange(DIRECT_EXCHANGE_NAME,true ,false);
    }

    @Bean
    Binding bingdingQueue02(){
        //将 queue02 和交换机绑定起来
        return BindingBuilder.bind(queue02())
                .to(directExchange())
        //为这个 Binding 设置一个 routingKey，将来消息发送的时候，会携带一个 routingKey，这个消息达到交换机之后，会自动将这个消息路由到与之同名
                .with(MY_QUEUE_NAME_02);
    }

    @Bean
    Binding bingdingQueue03(){
        //将 queue02 和交换机绑定起来
        return BindingBuilder.bind(queue03())
                .to(directExchange())
                .with(MY_QUEUE_NAME_03);
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46

    DirectConsumer

    监听两个消息队列

    @Component
public class DirectConsumer {
    private static final Logger logger = LoggerFactory.getLogger(DirectConsumer.class);

    @RabbitListener(queues = DirectConfig.MY_QUEUE_NAME_02)
    public void handleMsg(String msg){
        logger.info("{}:{}",DirectConfig.MY_QUEUE_NAME_02,msg);
    }

    @RabbitListener(queues = DirectConfig.MY_QUEUE_NAME_03)
    public  void handleMsg2(String msg){
        logger.info("{}:{}",DirectConfig.MY_QUEUE_NAME_03,msg);
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14

    生产者依旧使用第三小节中的producer01

    在producer01的测试类中加入方法：让其同时给customer02中的queue02和queue03发送数据如下：

         /**
     * 第一个参数：交换机
     * 第二个参数： routingKey ---》为了方便一般都是路径的名字
     * 第三个参数：生产者要发送的数据
     */
    @Test
    void test01() {
        rabbitTemplate.convertAndSend(DirectConfig.DIRECT_EXCHANGE_NAME, DirectConfig.MY_QUEUE_NAME_02, "hello queue02");
        rabbitTemplate.convertAndSend(DirectConfig.DIRECT_EXCHANGE_NAME, DirectConfig.MY_QUEUE_NAME_03, "hello queue03");
    }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10

    测试结果为
    queue02和queue03同时收到了消息，且消息不同，说明两个消费者消费的是不同的队列。今天的不同于昨天的是，今天是每个消费者消费自己的队列，且可控。
    在这里插入图片描述

    1.3.2 Fanout扇形交换机

    FanoutExchange 的数据交换策略是把所有到达 FanoutExchange 的消息转发给所有与它绑定的 Queue 上，在这种策略中，routingkey 将不起任何作用，

    fanout类型交换机会将接收到的消息广播给所有与之绑定的队列，也就是和路由键没有很大关系了
    1.消息到达交换机之后，会把消息自动转发给与扇形交换机绑定的所有队列

    在这里插入图片描述

    FanoutConfig步骤与Direct直连交换机基本相同

    1.定义所有队列和扇形交换机的名字
    2.把扇形交换机和队列注入Bean中
    3.绑定方法注入bean

    @Configuration
public class FanoutConfig {

    //1.定义所有队列和扇形交换机的名字
    public  static final  String MY_FANOUT_QUEUE_NAME_01 ="my fanout queue name 01";
    public  static final  String MY_FANOUT_QUEUE_NAME_02 ="my fanout queue name 02";
    public  static final  String MY_FANOUT_EXCHANGE_NAME ="my_fanout_exchange_name";

    //2.把扇形交换机和队列注入Bean中
    @Bean
    Queue fanoutQueue01(){
        return new Queue(MY_FANOUT_QUEUE_NAME_01,true,false,false);
    }

    @Bean
    Queue fanoutQueue02(){
        return new Queue(MY_FANOUT_QUEUE_NAME_02,true,false,false);
    }

    @Bean
    FanoutExchange fanoutExchange(){
        return new FanoutExchange(MY_FANOUT_EXCHANGE_NAME, true, false);
    }

    //3.绑定方法注入bean
    @Bean
    Binding fanoutBind01(){
        return BindingBuilder.bind(fanoutQueue01())
                .to(fanoutExchange());
    }

    @Bean
    Binding fanoutBind02(){
        return BindingBuilder.bind(fanoutQueue02())
                .to(fanoutExchange());
    }

}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39

    FanoutConsumer也基本相同

    
@Component
public class FanoutConsumer {

    private static final Logger logger = LoggerFactory.getLogger(FanoutConsumer.class);

    @RabbitListener(queues = FanoutConfig.MY_FANOUT_QUEUE_NAME_01)
    public void handleMsg(String msg) {
        logger.info("{}:{}", FanoutConfig.MY_FANOUT_QUEUE_NAME_01, msg);
    }

    @RabbitListener(queues = FanoutConfig.MY_FANOUT_QUEUE_NAME_02)
    public void handleMsg2(String msg) {
        logger.info("{}:{}", FanoutConfig.MY_FANOUT_QUEUE_NAME_02, msg);
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16

    生产者的测试类
    不同点:

    fanout类型交换机会将接收到的消息广播给所有与之绑定的队列，
也就是和路由键没有很大关系了,故此在这种策略中，routingkey
 将不起任何作用可以随意的填写。

    • 1
    • 2
    • 3
    @Test
    void test02() {
        rabbitTemplate.convertAndSend(FanoutConfig.MY_FANOUT_EXCHANGE_NAME, null, "hello fanout!");
    }

    • 1
    • 2
    • 3
    • 4

    结果：
    在这里插入图片描述

    1.3.3 Topic主机交换机（用的很多）

    TopicExchange 是比较复杂但是也比较灵活的一种路由策略，在 TopicExchange 中，Queue 通过 routingkey 绑定到 TopicExchange 上，当消息到达 TopicExchange 后，TopicExchange 根据消息的 routingkey 将消息路由到一个或者多个 Queue 上。

    一个生产者，一个交换机，两个队列，两个消费者在这里插入图片描述

    配置类 TopicConfig

    1.每一个交换机都有一个主题，消息队列可以订阅不同的主题
    2.# 表示通配符，将来发送消息的时候，如果消息的 routingKey 是 xiaomi.xxx，那么这个消息就会被转发到 xiaomiNewsQueue 队列上
    3.

    /**
 * 主题交换机
 * 每一个交换机都有一个主题，消息队列可以订阅不同的主题
 */
@Configuration
public class TopicConfig {
    public static final String XIAOMI_NEWS_QUEUE_NAME = "xiaomi_news_queue_name_01";
    public static final String HUAWEI_NEWS_QUEUE_NAME = "huawei_news_queue_name_01";
    public static final String PHONE_NEWS_QUEUE_NAME = "phone_news_queue_name_01";
    public static final String TOPIC_EXCHANGE_NAME = "topic_exchange_name";

    @Bean
    Queue xiaomiNewsQueue() {
        return new Queue(XIAOMI_NEWS_QUEUE_NAME, true, false, false);
    }

    @Bean
    Queue huaweiNewsQueue() {
        return new Queue(HUAWEI_NEWS_QUEUE_NAME, true, false, false);
    }

    @Bean
    Queue phoneNewsQueue() {
        return new Queue(PHONE_NEWS_QUEUE_NAME, true, false, false);
    }

    @Bean
    TopicExchange topicExchange() {
        return new TopicExchange(TOPIC_EXCHANGE_NAME, true, false);
    }

    @Bean
    Binding xiaomiBinding() {
        return BindingBuilder.bind(xiaomiNewsQueue())
                .to(topicExchange())
                //# 表示通配符，将来发送消息的时候，如果消息的 routingKey 是 xiaomi.xxx，那么这个消息就会被转发到 xiaomiNewsQueue 队列上
                .with("xiaomi.#");
    }

    @Bean
    Binding huaweiBinding() {
        return BindingBuilder.bind(huaweiNewsQueue())
                .to(topicExchange())
                .with("huawei.#");
    }

    @Bean
    Binding phoneBinding() {
        return BindingBuilder.bind(phoneNewsQueue())
                .to(topicExchange())
                .with("#.phone.#");
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    Rabbitmq延迟队列实现定时任务的方法
    pdf 0星 超过10%的资源 242KB
    下载

    实体类TopicConsumer

    @Component
public class TopicConsumer {
    private static final Logger logger = LoggerFactory.getLogger(TopicConsumer.class);

    @RabbitListener(queues = TopicConfig.XIAOMI_NEWS_QUEUE_NAME)
    public void handleMsg(String msg) {
        logger.info("{}:{}", TopicConfig.XIAOMI_NEWS_QUEUE_NAME, msg);
    }

    @RabbitListener(queues = TopicConfig.HUAWEI_NEWS_QUEUE_NAME)
    public void handleMsg2(String msg) {
        logger.info("{}:{}", TopicConfig.HUAWEI_NEWS_QUEUE_NAME, msg);
    }

    @RabbitListener(queues = TopicConfig.PHONE_NEWS_QUEUE_NAME)
    public void handleMsg3(String msg) {
        logger.info("{}:{}", TopicConfig.PHONE_NEWS_QUEUE_NAME, msg);
    }
}


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20

    生产者的测试类

    注意：
    1.第一个参数还是交换机的名字
    2.第二个参数是你的routingkey
    3.第三个参数是你要传入的信息

      @Test
    void test03() {
        //将被转发到xaiomi 的队列中
        rabbitTemplate.convertAndSend(TopicConfig.TOPIC_EXCHANGE_NAME, "xiaomi.news", "小米新闻");
        //将被转发到 huawei 的队列中
        rabbitTemplate.convertAndSend(TopicConfig.TOPIC_EXCHANGE_NAME, "huawei.news", "华为新闻");
        //将被转发到 xiaomi+phone 两个队列中
        rabbitTemplate.convertAndSend(TopicConfig.TOPIC_EXCHANGE_NAME, "xiaomi.phone", "小米手机新闻");

    }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10

    测试结果
    在这里插入图片描述

    1.3.4 Header头部交换机（用的少）

    HeadersExchange 是一种使用较少的路由策略，HeadersExchange 会根据消息的 Header 将消息路由 到不同的 Queue 上，这种策略也和 routingkey无关

    在这里插入图片描述

    配置类HeaderConfig

    @Configuration
public class HeaderConfig {

    public static final String HEADER_QUEUE_NAME01 = "header_queue_name01";
    public static final String HEADER_QUEUE_NAME02 = "header_queue_name02";
    public static final String HEADER_EXCHANGE_NAME = "header_exchange_name";

    @Bean
    Queue headerQueue01() {
        return new Queue(HEADER_QUEUE_NAME01, true, false, false);
    }

    @Bean
    Queue headerQueue02() {
        return new Queue(HEADER_QUEUE_NAME02, true, false, false);
    }

    @Bean
    HeadersExchange headersExchange() {
        return new HeadersExchange(HEADER_EXCHANGE_NAME, true, false);
    }

    @Bean
    Binding headerBinding01() {
        return BindingBuilder.bind(headerQueue01())
                .to(headersExchange())
                //如果将来的消息，消息头中有name 字段，就会进入到 headerQueue01 队列中
                .where("name").exists();
    }

    @Bean
    Binding headerBinding02() {
        return BindingBuilder.bind(headerQueue02())
                .to(headersExchange())
                //如果将来的消息，消息头中有age 字段且值为99，就会进入到 headerQueue02 队列中
                .where("age").matches(99);
    }

}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39

    HeadConsumer

    @Configuration
public class HeadConsumer {
    private static final Logger logger = LoggerFactory.getLogger(HeadConsumer.class);

    @RabbitListener(queues = HeaderConfig.HEADER_QUEUE_NAME01)
    public void handleMsg(byte[] msg) {
        logger.info("{}:{}", HeaderConfig.HEADER_QUEUE_NAME01, new  String(msg,0,msg.length));
    }

    @RabbitListener(queues = HeaderConfig.HEADER_QUEUE_NAME02)
    public void handleMsg2(byte[] msg) {
        logger.info("{}:{}", HeaderConfig.HEADER_QUEUE_NAME02, new  String(msg,0,msg.length));
    }      
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14

    生产者测试类

        @Test
    void test04() {
        //构建消息，并设置消息头
        Message msg01 = MessageBuilder.withBody("hello name".getBytes()).setHeader("name", "zhangsan").build();
        Message msg02 = MessageBuilder.withBody("hello age".getBytes()).setHeader("age", 99).build();
        rabbitTemplate.send(HeaderConfig.HEADER_EXCHANGE_NAME,null, msg01);
        rabbitTemplate.send(HeaderConfig.HEADER_EXCHANGE_NAME,null, msg02);
    }



    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10

    测试结果
    在这里插入图片描述

    1.4 Routing

    一个生产者，一个交换机，两个队列，两个消费者
    与1.3.1中的Direct直连十分相似
    在这里插入图片描述

    1.5 Topics

    一个生产者，一个交换机，两个队列，两个消费者，生产者创建 Topic 的Exchange 并且绑定到队列中，这次绑定可以通过 * 和 # 关键字，对指定 RoutingKey 内容，编写时注意格式 xxx.xxx.xxx去编写。
    在这里插入图片描述
    与1.3.1中的 主机交换Topic十分相似

    1.6 RPC

    什么是RPC：（Remote Procedure Call Protocol 远程过程调用协议），后面的跨进程调用都是用到了这个

    1.6.1 消息中间件怎么实现解耦的功能？

    1.a调用b的时候b必须在线不在线就失败了
    2.我们可以用消息中间件解决，a调用b就是想用b的服务。
    具体过程是：.
    a给消息中间件发送消息，b去消费那个消息，消费完了之后，b也给消息中间件发送一个消息，然后a再去消费那个消息，这样就通过消息中间件完成了解耦
    就是a把消息扔到消息中间件后就可以不用管了，如果b在线立马就会消费，如果b不在线那个消息本身也不会丢失，等b上线了就会把a丢在消息中间件的消息给消费了。

    RPC调用说的基本上就是这么个情况

    1.6.2 简单架构图

    在这里插入图片描述

    1.首先Client发送一条消息，和普通的消息相比，这条消息多了两个关键内容：
    correlation_id,这个表示这条消息的唯一id，还有一个内容是reply_to,这个表示消息回复队列的名字
    Server从消息发送队列获取消息并且处理响应的业务逻辑，处理完成之后，再将处理结果发送到reply_to指定的回调队列中。
    Client从回调队列中读取消息，就可以知道消息的执行情况是什么样子了

    1.6.3 RPC 调用的小例子

    rabbitmq-demo.rar
    rar 0星 超过10%的资源 17KB
    下载

    1.6.3.1 建maven

    我们先创建一个客户端：rpc_client
    再创建一个服务端：rpc_server

    1.6.3.2 依赖

    rpc_client

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

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

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19

    rpc_server

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

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

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19

    1.6.3.3application配置

    rpc_client

    spring.rabbitmq.host=1.12.235.192
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/

#添加了这个属性之后，将来发送的消息会自动携带一个id,通过这个id我们可以识别出自己想要的消息
spring.rabbitmq.publisher-confirm-type=correlated
spring.rabbitmq.publisher-returns=true

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9

    rpc_server

    1.配置correlation_id，将发送的消息和返回值之间关联起来
    2.开启发送失败退回

    spring.rabbitmq.host=1.12.235.192
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/

server.port=8085
#只有开启了这个配置，将来的消息中才会带 correlation_id，只有通过correlation_id 我们才能将发送的消息和返回值之间关联起来
spring.rabbitmq.publisher-confirm-type=correlated
#是开启发送失败退回。
spring.rabbitmq.publisher-returns=true

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11

    1.6.3.4 设置配置配RpcConfig

    rpc_client 中的配置类RpcConfig

    关键点：
    1.消息发送调用 sendAndReceive 方法，该方法自带返回值，返回值就是服务端返回的消息。
    2.服务端返回的消息中，头信息中包含了 spring_returned_message_correlation 字段，这个就是消息发送时候的 correlation_id，通过消息发送时候correlation_id 以及返回消息头中的 spring_returned_message_correlation 字段值，我们就可以将返回的消息内容和发送的消息绑定到一起，确认出这个返回的内容就是针对这个发送的消息的。
    3.客户端的开发最核心的就是sendAndReceive方法的调用。主要点就是发送的消息中要有correlation_id即我们的application配置类中要有其配置。

    /**
 * client 发送消息，发到 RPC_MSG_QUEUE 队列，server 监听这个队列，就可以收到 client 发来的消息了，然后 server 对消息进行处理，处理完成后，将响应的内容发到 RPC_REPLY_QUEUE 队列上，client 监听 RPC_REPLY_QUEUE 队列，就可以收到 server 返回的信息了
 */
@Configuration
public class RpcConfig {
    //client 发送消息的时候，发到这个队列上（server 将来监听这个队列）
    public static final String RPC_MSG_QUEUE = "rpc_msg_queue";
    //server 响应消息的时候，发送到这个队列上（client 将来监听这个队列）
    public static final String RPC_REPLY_QUEUE = "rpc_reply_queue";
    public static final String RPC_EXCHANGE_NAME = "rpc_exchange_name";

    @Bean
    Queue rpcMsgQueue() {
        return new Queue(RPC_MSG_QUEUE, true, false, false);
    }

    @Bean
    Queue rpcReplyQueue() {
        return new Queue(RPC_REPLY_QUEUE, true, false, false);
    }

    @Bean
    TopicExchange topicExchange() {
        return new TopicExchange(RPC_EXCHANGE_NAME, true, false);
    }

    @Bean
    Binding rpcReplyBinding() {
        return BindingBuilder.bind(rpcReplyQueue())
                .to(topicExchange())
                .with(RPC_REPLY_QUEUE);
    }

    @Bean
    Binding rpcMsgBinding() {
        return BindingBuilder.bind(rpcMsgQueue())
                .to(topicExchange())
                .with(RPC_MSG_QUEUE);
    }

    @Bean
    RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        //设置响应队列名
        rabbitTemplate.setReplyAddress(RPC_REPLY_QUEUE);
        //设置响应超时时间
        rabbitTemplate.setReplyTimeout(6000);
        return rabbitTemplate;
    }

    /**
     * 当 server 响应数据的时候，数据是直接发送到队列中的，按照旧的方法就应该写一个消费者，但是这样会将消息的发送和接受分裂开。
     *
     * 所以，我们配置这个监听器，就可以实现消息的发送和接受在一起
     * @param connectionFactory
     * @return
     */
    @Bean
    SimpleMessageListenerContainer simpleMessageListenerContainer(ConnectionFactory connectionFactory) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setQueueNames(RPC_REPLY_QUEUE);
        container.setMessageListener(rabbitTemplate(connectionFactory));
        return container;
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65

    rpc_server

    1.这里的关键是配置一下消息队列并且把消息队列和消息交换机绑定起来

    /**
 * client 发送消息，发到 RPC_MSG_QUEUE 队列，server 监听这个队列，就可以收到 client 发来的消息了，然后 server 对消息进行处理，处理完成后，将响应的内容发到 RPC_REPLY_QUEUE 队列上，client 监听 RPC_REPLY_QUEUE 队列，就可以收到 server 返回的信息了
 */
@Configuration
public class RpcConfig {

    //client 发送消息的时候，发到这个队列上（server 将来监听这个队列）
    public static final String RPC_MSG_QUEUE = "rpc_msg_queue";
    //server 响应消息的时候，发送到这个队列上（client 将来监听这个队列）
    public static final String RPC_REPLY_QUEUE = "rpc_reply_queue";
    public static final String RPC_EXCHANGE_NAME = "rpc_exchange_name";

    @Bean
    Queue rpcMsgQueue() {
        return new Queue(RPC_MSG_QUEUE, true, false, false);
    }

    @Bean
    Queue rpcReplyQueue() {
        return new Queue(RPC_REPLY_QUEUE, true, false, false);
    }

    @Bean
    TopicExchange topicExchange() {
        return new TopicExchange(RPC_EXCHANGE_NAME, true, false);
    }

    @Bean
    Binding rpcReplyBinding() {
        return BindingBuilder.bind(rpcReplyQueue())
                .to(topicExchange())
                .with(RPC_REPLY_QUEUE);
    }

    @Bean
    Binding rpcMsgBinding() {
        return BindingBuilder.bind(rpcMsgQueue())
                .to(topicExchange())
                .with(RPC_MSG_QUEUE);
    }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40

    1.6.3.4 配置Controller

    rpc_client 中的RpcClientController

    @RestController
public class RpcClientController {

    private static final Logger logger = LoggerFactory.getLogger(RpcClientController.class);

    @Autowired
    RabbitTemplate rabbitTemplate;

    @GetMapping("/send")
    public void sendMsg(String msg) {
        Message message = MessageBuilder.withBody(msg.getBytes()).build();
        //这里拿到的响应值，就是服务端返回的消息
        Message result = rabbitTemplate.sendAndReceive(RpcConfig.RPC_EXCHANGE_NAME, RpcConfig.RPC_MSG_QUEUE, message);
        if (result != null) {
            //现在判断一下这个响应的数据，是不是刚刚发送的请求的响应
            //这个是刚刚发送的消息自动携带的 id
            String correlationId = message.getMessageProperties().getCorrelationId();
            String result_correlationId = (String) result.getMessageProperties().getHeaders().get("spring_returned_message_correlation");
            if (correlationId.equals(result_correlationId)) {
                //说明响应的这个消息就是刚刚发送消息的响应
                byte[] body = result.getBody();
                logger.info("服务端返回数据：{}", new String(body, 0, body.length));
            }
        }
    }

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25

    rpc_server中的RpcServerController

    这里的逻辑

    1. 服务端首先收到消息并打印出来。
    2. 服务端提取出原消息中的 correlation_id。
      3.服务端利用sendAndReceive方法，将消息发送给RPC_EXCHANGE_NAME和RPC_REPLY_QUEUE队列，同时带上correlationId参数
      4.服务端的消息发出后，客户端会受到服务端返回的结果
    @RestController
public class RpcServerController {

    private static final Logger logger = LoggerFactory.getLogger(RpcServerController.class);

    @Autowired
    RabbitTemplate rabbitTemplate;

    @RabbitListener(queues = RpcConfig.RPC_MSG_QUEUE)
    public void handleMsg(Message message) {
        byte[] body = message.getBody();
        //打印出来信息
        logger.info("客户端发送来消息:{}", new String(body, 0, body.length));
        //获取客户端发送来的消息的 correlationId
        String correlationId = message.getMessageProperties().getCorrelationId();
        Message msg = MessageBuilder.withBody("收到消息".getBytes()).build();
        //服务端利用sendAndReceive方法，将消息发送给RPC_EXCHANGE_NAME和RPC_REPLY_QUEUE队列，同时带上correlationId参数
        rabbitTemplate.sendAndReceive(RpcConfig.RPC_EXCHANGE_NAME,RpcConfig.RPC_REPLY_QUEUE,msg,new CorrelationData(correlationId));
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20

    测试

    1.7 Publisher Confirms

  • 相关阅读:
    基于SSM的健身房管理系统
    Go语言的流程控制
    数学建模美赛入门
    vncserver 安装、自启动、安全
    5-2.Binding指定源
    c++踩坑点，类型转换
    Postgresql RECORD与%ROWTYPE类型
    爬虫基础 - 爬虫学的好，牢饭吃得饱系列
    Linux创建YUM仓库
    某卢小说网站登录密码逆向
  • 原文地址：https://blog.csdn.net/weixin_43189971/article/details/126256175