SpringBoot整合RabbitMQ (持续更新中)

发布于:2024-06-11 ⋅ 阅读:(13) ⋅ 点赞:(0)

RabbitMQ 官网地址:RabbitMQ: One broker to queue them all | RabbitMQ

RabbitMQ 与 Erlang 版本兼容关系

  • 3.13.0
  • 26.0
  • 26.2.x
  • The 3.13 release series is compatible with Erlang 26.

  • OpenSSL 3 support in Erlang is considered to be mature and ready for production use.

  • Erlang 26.1 and later versions supports FIPS mode on OpenSSL 3

  • 3.12.13
  • 3.12.12
  • 3.12.11
  • 3.12.10
  • 25.0
  • 26.2.x
  • The 3.12 release series is compatible with Erlang 26.

  • OpenSSL 3 support in Erlang is considered to be mature enough for production.

  • Erlang 26.1 and later versions supports FIPS mode on OpenSSL 3

  • 3.12.9
  • 3.12.8
  • 3.12.7
  • 3.12.6
  • 3.12.5
  • 25.0
  • 26.1.x
  • The 3.12 release series is compatible with Erlang 26.

  • OpenSSL 3 support in Erlang is considered to be mature enough for production.

  • Erlang 26.1 supports FIPS mode on OpenSSL 3

  • 3.12.4
  • 3.12.3
  • 3.12.2
  • 3.12.1
  • 3.12.0
  • 25.0
  • 26.0.x
  • The 3.12 release series is compatible with Erlang 26.

  • OpenSSL 3 support in Erlang is considered to be mature enough for production.

  • 3.11.28
  • 3.11.27
  • 3.11.26
  • 3.11.25
  • 3.11.24
  • 3.11.23
  • 3.11.22
  • 3.11.21
  • 3.11.20
  • 3.11.19
  • 3.11.18
  • 3.11.17
  • 3.11.16
  • 3.11.15
  • 3.11.14
  • 3.11.13
  • 3.11.12
  • 3.11.11
  • 3.11.10
  • 3.11.9
  • 3.11.8
  • 3.11.7
  • 3.11.6
  • 3.11.5
  • 3.11.4
  • 3.11.3
  • 3.11.2
  • 3.11.1
  • 3.11.0
  • 25.0
  • 25.3.x
  • Erlang 26 is supported starting with RabbitMQ 3.12.0.

  • As of Erlang 25.1, OpenSSL 3.0 support in Erlang is considered to be mature enough for production.

  • Erlang 25 before 25.0.2 is affected by CVE-2022-37026, a CVE with critical severity (CVSS 3.x Base Score: 9.8)

RabbitMQ 安装

下载地址:RabbitMQ: One broker to queue them all | RabbitMQ

exe文件点击安装即可(其他系统版本看官网)

RabbitMQ管理界面

管理界面的默认端口:15672 ,默认账户/密码: gurest/guest

SpringBoot整合RabbitMQ

1.maven 依赖

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

2.添加配置文件

spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

3.添加配置文件

package com.label.config;

import org.springframework.amqp.core.*;

import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

/**
 * @author sszdzq
 */
@Component
public class RabbitMqConfig {

    /**
     * 定义一个主题类型的交换机
     *
     * @return
     */
    @Bean("topic-exchange")
    public Exchange topicExchange() {
        return ExchangeBuilder
                .topicExchange("topic-exchange") // 交换机类型,交换机名称
                .durable(true) //ture为持久化,存到磁盘,false存到内存
                .build();
    }

    /**
     * 定义一个队列
     *
     * @return
     */
    @Bean("messageQueue")
    public Queue messageQueue() {
        return new Queue("messageQueue");
    }

    /**
     * 交换机、队列、路由键 进行绑定
     *
     * @param exchange //交换机
     * @param queue //队列
     * @return
     */
    @Bean
    public Binding bindQueueAndExchange(@Qualifier("topic-exchange") Exchange exchange, @Qualifier("messageQueue") Queue queue) {
        return BindingBuilder
                .bind(queue)
                .to(exchange)
                .with("news.*") //路由键
                .noargs();
    }
}

4.创建生产者与消费者

package com.label.contoller;

import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.json.JSONException;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.io.IOException;

@RestController
@Slf4j
@RequestMapping(value = "/rabbitmq")
public class TestController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 直连类型
     *
     * @param msg
     * @return
     */
    @PostMapping("/directSend")
    public ResponseEntity sendMsgDirect(String msg) {
        /**
         * 普通直接链接
         */
        rabbitTemplate.convertAndSend("direct_exchange", "direct_key", msg);
        /**
         * 设置属性 发送
         */
        rabbitTemplate.convertAndSend("direct_exchange", "direct_key", msg, e -> {
            MessageProperties messageProperties = e.getMessageProperties();
            /**
             * 单位为毫秒("6000",6秒钟)
             * 过期后如果设置了死信队列,消息进入死信队列
             * 没有设置死信直接丢弃
             */
            messageProperties.setExpiration("6000");
            return e;
        });
        return ResponseEntity.ok("3482347592");
    }

    /**
     * 扇形消息发送
     *
     * @param msg
     * @return
     */
    @PostMapping("/fanoutSend")
    public ResponseEntity sendMsgFanout(String msg) {
        /**
         * 广播模式下 没有路由建信息(填写也是无效)
         */
        rabbitTemplate.convertAndSend("fanout_exchange", "", "this is test message");
        return ResponseEntity.ok("3482347592");
    }

    /**
     * 广播模式消费
     *
     * @param msg
     */
    @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(value = "fanout_exchange", type = ExchangeTypes.FANOUT),
            value = @Queue("fanout_queue")))
    public void customerFanout(String msg) {
        log.info("广播消费:{}", msg);
    }


    /**
     * 类型:主题
     * 生产者 (创建消息)
     * 主题消费发送
     */
    @PostMapping("/topic/send")
    public ResponseEntity producerTopicSend(@RequestBody JSONObject jb) throws JSONException {
        rabbitTemplate.convertAndSend("topic-exchange", jb.getString("topic"), jb.getString("msg"));
        return ResponseEntity.ok().build();
    }

    /**
     * 创建新的队列(绑定交换机:news.*,绑定路由键:news.330500000000)
     * x-expires: 队列的销毁时间
     *
     * @param msg
     */
    @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(value = "topic-exchange", type = ExchangeTypes.TOPIC),
            value = @Queue(value = "message-one7", arguments = {
                    @Argument(name = "x-expires", value = "10000", type = "java.lang.Integer")
            }),
            key = {"news.330500000000"}
    ))
    public void customerTopic1(String msg) {
        log.info("主题消费 news.330500000000 :{}", msg);
    }

    /**
     * 手动确认
     *
     * @param message
     * @param channel
     * @throws IOException
     */
    @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(value = "topic-exchange", type = ExchangeTypes.TOPIC),
            value = @Queue(value = "message-one"),
            key = {"news.*"}
    ))
    public void customerTopic2(Message message, Channel channel) throws IOException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            log.info("主题消费 news.* :{}", new String(message.getBody()));
            channel.basicAck(deliveryTag, true);
        } catch (Exception e) {
            channel.basicNack(deliveryTag, true, true);
            throw new RuntimeException(e);
        }
    }
}