微服务: 03-rabbitmq在springboot中如何使用(下篇)-Toy模板网

这篇具有很好参考价值的文章主要介绍了微服务: 03-rabbitmq在springboot中如何使用(下篇)。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

前言: 上文传送

4.六大模式实际操作(续)

4.4 路由模式:

---> 4.4.1 消费者配置类

---> 4.4.2 消费者代码

--->4.4.3 生产者代码

4.5 主题模式: (路由升级版)

---> 4.5.1 消费者配置类

---> 4.5.2 消费者代码

---> 4.5.3 生产者代码

---> 4.5.4 测试效果

4.6 RPC异步调用模式(用的少)

---> 4.6.0 找了个mq的简史还是蛮有趣的: 消息中间件的发展史是一个有趣的历史故事

---> 4.6.1 消费者配置类

---> 4.6.2 生产者配置类

---> 4.6.3 消费者代码

---> 4.6.4 生产者代码

---> 4.6.5 实现效果

文章总结:

前言: 上文传送

微服务 02-rabbitmq在springboot中如何使用(上篇)

4.六大模式实际操作(续)

4.4 路由模式:

---> 4.4.1 消费者配置类

package cn.pingzhuyan.rabbitmq.config;

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

/**
 * rabbitmq的默认手动确认模式
 * @author pzy
 * @version 0.1.0
 * @description: TODO
 */
@Configuration
public class RabbitConfig {

    /**
     * 创建(声明)一个简单队列
     * @return
     */
    @Bean
    public Queue helloQueue(){
//        return new Queue("PZY",true,false,false);
        return new Queue("PZY");
    }

    /**
     * 创建radioFanout交换机
     * 消费者需要绑定此交换机
     * @return
     */
    @Bean
    public FanoutExchange radioFanout(){
        return new FanoutExchange("PZY_RADIO",false,false);
    }

    /**
     * 路由模式 指定路由交换机
     * 指定接收路由键
     * @return
     */
    @Bean
    public DirectExchange directExchange(){
        return new DirectExchange("PZY_DIRECT",false,false);
    }

}

---> 4.4.2 消费者代码

package cn.pingzhuyan.rabbitmq.directM4;

import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * 路由交换机消费者
 *
 * @author pzy
 * @version 0.1.0
 * @description: TODO
 */
@Component
public class DirectM4Consumer {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue,  //不写就是随机队列, false true true
            exchange = @Exchange(name = "PZY_DIRECT", declare = "false"),//交换机(PZY_RADIO, 不创建并使用已经存在的交换机)
            key = {"pzy1", "pzy2"}
    ))
    public void radioFanoutMessage1(String msg) {
        System.out.printf("消费者1接收到<1/2>: %s\n", msg);
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue,  //不写就是随机队列, false true true
            exchange = @Exchange(name = "PZY_DIRECT", declare = "false"),//交换机(PZY_RADIO, 不创建并使用已经存在的交换机)
            key = {"pzy3", "pzy4"}
    ))
    public void radioFanoutMessage2(String msg) {
        System.out.printf("消费者2接收到<3/4>: %s\n", msg);
    }

}

--->4.4.3 生产者代码

package cn.pingzhuyan.rabbitmq.directM4;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Random;

/**
 * 路由交换机生产者
 *
 * @author pzy
 * @version 0.1.0
 * @description: TODO
 */
@Component
public class DirectM4Provider {

    @Autowired
    private AmqpTemplate amqpTemplate;


    @Async
    @Scheduled(cron = "*/1 * * * * ?")
    public void directSend01() {

        int num = new Random().nextInt(4) + 1;//生成1-4的随机数

        System.out.println("生产者1: <发布和订阅模式>定时(1次/s)发送 -> " + "我是消息,编号是: " + num);

        amqpTemplate.convertAndSend("PZY_DIRECT", "pzy" + num, "我是消息,编号是: " + num);
    }


}

4.5 主题模式: (路由升级版)

---> 4.5.1 消费者配置类

    /**
     * 主题模式(升级版路由) 指定路由交换机
     * 指定接收路由键表达式
     * @return
     */
    @Bean
    public TopicExchange topicExchange(){
        return new TopicExchange("PZY_TOPIC",false,false);
    }

---> 4.5.2 消费者代码

//测试四种

//1. pzy.pzy.pzy 1和2 都能收到

//2. aaa.pzy.bbb 1能收到

//3. pzy.aaa.bbb.ccc 2能收到

//4. aaa.bbb.pzy 2能收到

package cn.pingzhuyan.rabbitmq.topicM5;

import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * 路由交换机消费者
 * 测试四种
 * 1. pzy.pzy.pzy     1和2 都能收到
 * 2. aaa.pzy.bbb     1能收到
 * 3. pzy.aaa.bbb.ccc 2能收到
 * 4. aaa.bbb.pzy     2能收到
 *
 * @author pzy
 * @version 0.1.0
 * @description: TODO
 */
@Component
public class TopicM5Consumer {

    /**
     * 消费者1
     *
     * @param msg
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue,  //不写就是随机队列, false true true
            exchange = @Exchange(name = "PZY_TOPIC", declare = "false"),//交换机(PZY_RADIO, 不创建并使用已经存在的交换机)
            key = {"*.pzy.*"}
    ))
    public void radioFanoutMessage1(String msg) {
        System.out.printf("消费者1接收到: %s\n", msg);
    }

    /**
     * 消费者2
     *
     * @param msg
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue,  //不写就是随机队列, false true true
            exchange = @Exchange(name = "PZY_TOPIC", declare = "false"),//交换机(PZY_RADIO, 不创建并使用已经存在的交换机)
            key = {"pzy.#", "*.*.pzy"}
    ))
    public void radioFanoutMessage2(String msg) {
        System.out.printf("消费者2接收到: %s\n", msg);
    }

}

---> 4.5.3 生产者代码

@PostConstruct解释传送门===> : 启动类加载时方法执行的几种实现方式

package cn.pingzhuyan.rabbitmq.topicM5;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.Scanner;

/**
 * 主题模式生产者
 *
 * @author pzy
 * @version 0.1.0
 * @description: TODO
 */
@Component
public class TopicM5Provider {

    @Autowired
    private AmqpTemplate amqpTemplate;
    
    @PostConstruct
    public void topicSend01() {

        for (; ; ) {
            System.out.println("请输入路由键: ");
            String routingKey = new Scanner(System.in).nextLine();

            /*中断测试,启动程序完成*/
            if ("exit".equalsIgnoreCase(routingKey)) break;

            System.out.println("请输入消息: ");
            String message = new Scanner(System.in).nextLine();

            System.out.println("生产者1: <发布和订阅模式>发送 -> " + message + "路由键是: " + routingKey);

            amqpTemplate.convertAndSend("PZY_TOPIC", routingKey, message);
        }
    }
}

---> 4.5.4 测试效果

微服务: 03-rabbitmq在springboot中如何使用(下篇),# 分布式必备服务配置,java-rabbitmq,微服务,rabbitmq,spring boot,spring cloud

4.6 RPC异步调用模式(用的少)

[个人理解, 不喜勿喷]

有成熟的rpc 框架 dubbo Netflix(feign,ribbon,Hystrix), gRpc等功能更加完善,在微服务架构中服务与服务内部通信必不可少的技术.

使用rabbitmq的rpc模式性能瓶颈就在它身上, 如果mq宕机整个服务系统都不可用.
之后看mq的简史, 这个技术应该是在微服务诞生之前就出现了, 了解了这模式用的人少原因

---> 4.6.0 找了个mq的简史还是蛮有趣的: 消息中间件的发展史是一个有趣的历史故事

---> 4.6.1 消费者配置类

第二个随机队列可以不写声明在哪都行

   /**
     * RPC异步调用模式(用的最少)
     * @return
     */
    @Bean
    public Queue sendQueue() {
        return new Queue("PZY_PRC", false);
    }

    /**
     * 声明PRC一个随机队列  (生产者需要监听这个队列 收到回执消息)
     * @return
     */
    @Bean
    public Queue rpcRandomQueue() {
        return new Queue(UUID.randomUUID().toString(), false);
    }

---> 4.6.2 生产者配置类

需要获取到queue的名称 spring给了获取bean的方法 #{rpcRandomQueue.name}

package cn.pingzhuyan.rabbitmq.config;

import org.springframework.amqp.core.Queue;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.UUID;

/**
 * 生产者
 * rabbitmq的默认手动确认模式
 * <p>
 * 只需要声明一个随机队列 绑定生成的验证码
 *
 * @author pzy
 * @version 0.1.0
 * @description: TODO
 */
@Configuration
public class RabbitConfig {

    /**
     * 声明一个随机队列
     *
     * @return
     */
    @Bean
    public Queue rpcRandomQueue() {
        return new Queue(UUID.randomUUID().toString(), false);
    }

}

---> 4.6.3 消费者代码

业务代码随便, 主要是要这个业务处理结果, 写个相加, 随机数, 猜字符都行

package cn.pingzhuyan.rabbitmq.RpcM6;

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

import java.util.Random;

/**
 * Rpc模式 消费者
 *
 * 其实也是生产者 将收到的消息放入随机队列
 * @author pzy
 * @version 0.1.0
 * @description: TODO
 */
@Component
public class RpcM6Consumer {

    @RabbitListener(queues = "PZY_PRC")
    public String getAuthCode(int length) {

        String authCode = makeAuthCode(length);

        System.out.printf("由消费者1制作完成: %s \n", authCode);

        return authCode;
    }

    @RabbitListener(queues = "PZY_PRC")
    public String getAuthCode1(int length) {
        String authCode = makeAuthCode(length);

        System.out.printf("由消费者1制作完成: %s \n", authCode);

        return authCode;
    }

    /**
     * 制作验证码
     *
     * @param length
     * @return
     */
    private String makeAuthCode(int length) {
        StringBuilder builder = new StringBuilder();
        for (int i = 0; i < 26; i++) {
            builder.append((char) ('a' + i));
            builder.append((char) ('A' + i));

            if (i < 10) builder.append(i);

        }
        String line = builder.toString();
        StringBuilder b = new StringBuilder();
        Random rand = new Random();
        for (int i = 0; i < length; i++) {
            b.append(line.charAt(rand.nextInt(line.length())));
        }

        try {
            Thread.sleep(rand.nextInt(2000));//阻塞0-5000ms
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.printf("生成 %s 位数验证码, ", length);
        return b.toString();
    }


}

---> 4.6.4 生产者代码

package cn.pingzhuyan.rabbitmq.RpcM6;


import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Random;

import java.util.UUID;

/**
 * rpc模式 生产者
 * <p>
 * 也是消费者 从绑定的随机队列中接收到消息
 *
 * @author pzy
 * @version 0.1.0
 * @description: TODO
 */

@Component
@EnableAsync
@EnableScheduling
public class RpcM6Provider {

    @Autowired
    private AmqpTemplate amqpTemplate;


    @Value("#{rpcRandomQueue.name}")
    private String rpcRandomQueue;


    /**
     * 定时操作 生成随机位数的验证码
     */
    @Async
    @Scheduled(cron = "*/5 * * * * ?")
    public void topicSend01() {

        int length = new Random().nextInt(19) + 1;//自然数
        System.out.printf("验证码生成器(随机生成位数): %s \n", length);
//        if (length <= 0) length = 1;

        amqpTemplate.convertAndSend("PZY_PRC", length, message -> {
            MessageProperties messageProperties = message.getMessageProperties();
            messageProperties.setReplyTo(rpcRandomQueue);
            messageProperties.setCorrelationId(UUID.randomUUID().toString());
            return message;
        });
    }

    /**
     * 生产者-> 消费者  接收绑定了消息的随机队列
     * 然后进行后面逻辑业务
     * @param authCode
     * @param correlationId
     */
    @RabbitListener(queues = "#{rpcRandomQueue.name}")
    public void receive(String authCode, @Header(name = AmqpHeaders.CORRELATION_ID) String correlationId) {

        System.out.println("生产者: 收到关联id号-> " + correlationId + "的验证码是: " + authCode);
    }


}