消息中间件RabbitQM

Posted 2023-03-29 leese233

作用：消息中间件、异步通讯
Spring AMQP封装了RabbitMQ，简化了消息发送和接收的操作

简单消息的发送
@Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    public void testSimpleQueue()
        // 队列名称
        String queueName = "simple.queue";
        // 消息
        String message = "hello, spring amqp!";
        // 发送消息
        rabbitTemplate.convertAndSend(queueName, message);
   

简单消息的接收
 @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueueMessage(String msg) throws InterruptedException
        System.out.println("spring 消费者接收到消息：【" + msg + "】");
   

Fanout广播模式
可以有多个队列
每个队列都要绑定到Exchange（交换机）
生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定
交换机把消息发送给绑定过的所有队列
订阅队列的消费者都能拿到消息
 /**
     * 声明交换机
     * @return Fanout类型交换机
     */
    @Bean
    public FanoutExchange fanoutExchange()
        return new FanoutExchange("itcast.fanout");
   

    /**
     * 第1个队列
     */
    @Bean
    public Queue fanoutQueue1()
        return new Queue("fanout.queue1");
   

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange)
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
   

发送消息到交换机
@Test
public void testFanoutExchange()
    // 队列名称
    String exchangeName = "itcast.fanout";
    // 消息
    String message = "hello, everyone!";
    rabbitTemplate.convertAndSend(exchangeName, "", message);

Direct模式
注解方式声明
队列和交换机绑定
@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "direct.queue1"),
    exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
    key = "red", "blue"
))
public void listenDirectQueue1(String msg)
    System.out.println("消费者接收到direct.queue1的消息：【" + msg + "】");

消息发送
@Test
public void testSendDirectExchange()
    // 交换机名称
    String exchangeName = "itcast.direct";
    // 消息
    String message = "红色警报！日本乱排核废水，导致海洋生物变异，惊现哥斯拉！";
    // 发送消息
    rabbitTemplate.convertAndSend(exchangeName, "red", message);

Topic模式
队列和交换机绑定
@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "topic.queue1"),
    exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),
    key = "china.#"
))
public void listenTopicQueue1(String msg)
    System.out.println("消费者接收到topic.queue1的消息：【" + msg + "】");

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "topic.queue2"),
    exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),
    key = "#.news"
))
public void listenTopicQueue2(String msg)
    System.out.println("消费者接收到topic.queue2的消息：【" + msg + "】");

发送消息
@Test
public void testSendTopicExchange()
    // 交换机名称
    String exchangeName = "itcast.topic";
    // 消息
    String message = "喜报！孙悟空大战哥斯拉，胜!";
    // 发送消息
    rabbitTemplate.convertAndSend(exchangeName, "china.news", message);

RabbitQM秒杀源码

• 高并发商品限时秒杀系统JAVA源码

https://github.com/zaiyunduan123

https://github.com/zaiyunduan123/springboot-seckill