欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

SpringBoot整合rabbitMQ,spring-boot-starter-amqp 的使用

程序员文章站 2022-07-12 12:28:58
...

在springBoot项目中使用rabbitMQ是很方便的,spring提供了spring-boot-starter-amqp依赖,只需要简单的配置即可与spring无缝整合

本文不介绍rabbitMQ理论知识,主要介绍在spring中如何使用rabbirMQ

依赖与配置
maven

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

配置文件

spring:
  rabbitmq:
    username: zolvces
    password: zolvces
    addresses: 192.168.39.128:5672
    cache:
      connection:
#       缓存连接模式,默认一个连接,多个channel
        mode: channel
#        多个连接,多个channel
#        mode: connection

几种交换机
Direct 直连交换机
申明一个消费者,申明一个queue和Exchange(如果他们不存在),并binding

在这里插入代码片@Component
@RabbitListener(bindings = @QueueBinding(
        value = @Queue("myDirectQueue"),
        exchange = @Exchange(value = "myDirectExchange", type = ExchangeTypes.DIRECT),
        key = "mine.direct"
))
public class MyDirectListener {

    /**
     * listenerAdapter
     *
     * @param msg 消息内容,当只有一个参数的时候可以不加@Payload注解
     */
    @RabbitHandler
    public void onMessage(@Payload String msg) {
        System.out.println(msg);
    }
}

生产者

    @Autowired
    private RabbitTemplate rabbitTemplate;

    private void send() {
        //(交换机,routingKey,消息内容)
        rabbitTemplate.convertAndSend("myDirectExchange","mine.direct","this is a message");
    }

Default 默认交换机
默认交换机其实就是直连交换机,可以理解为名称为空字符串的直连交换机,一个queue若不指定binding的交换机,就被绑定到默认交换机上,routingKey为queue的名称

消费者

@Component
@RabbitListener(queuesToDeclare = @Queue("myDefaultQueue"))
public class MyDefaultListener {

    @RabbitHandler
    public void onMessage(String msg) {
        System.out.println(msg);
    }
}

生产者

  @Autowired
    private RabbitTemplate rabbitTemplate;
    
    private void send() {
        //(队列,消息内容)
        rabbitTemplate.convertAndSend("myDefaultQueue","this is a message");
    }
Fanout 扇型交换机
申明两个消费者,对应queue-one和queue-two 都与 myFanoutExchange绑定,因为是扇形交换机,所以这里的routingKey是随意的

@Component
@RabbitListeners({
    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue("myFanoutQueue-one"),
                    exchange = @Exchange(value = "myFanoutExchange", type = ExchangeTypes.FANOUT),
                    key = "key.one")),
    
    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue("myFanoutQueue-two"),
                    exchange = @Exchange(value = "myFanoutExchange", type = ExchangeTypes.FANOUT),
                    key = "key.two")),
})
public class MyFanoutListener {

    @RabbitHandler
    public void onMessage(@Payload String msg, @Headers Map<String, Object> headers) {
        System.out.println("来自" + headers.get(AmqpHeaders.CONSUMER_QUEUE) + "的消息:" + msg);
    }

}

生产者

 @Autowired
    private RabbitTemplate rabbitTemplate;

    private void send() {
        //(交换机,routingKey,消息内容),routingKey随意
        rabbitTemplate.convertAndSend("myFanoutExchange","key.one","this is a message");
    }

Topic 主题交换机
申明多个消费者,假设场景,有人在**商店中奖了,各个级别的电视台进行新闻报道.

通配符 # : 一个或多个word, * 一个word

@Component
public class MyTopicListener {

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("province-news-queue"),
            exchange = @Exchange(value = "news-exchange", type = ExchangeTypes.TOPIC),
            key = "province.#"))
    @RabbitHandler
    public void provinceNews(String msg) {
        System.out.println("来自省TV的消息:" + msg);
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("city-news-queue"),
            exchange = @Exchange(value = "news-exchange", type = ExchangeTypes.TOPIC),
            key = "province.city.#"))
    @RabbitHandler
    public void cityNews(String msg) {
        System.out.println("来自市TV的消息:" + msg);
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("street-news-queue"),
            exchange = @Exchange(value = "news-exchange", type = ExchangeTypes.TOPIC),
            key = "province.city.street.*"))
    @RabbitHandler
    public void streetNews(String msg) {
        System.out.println("来自街区TV的消息:" + msg);
    }

}

生产者

  @Autowired
    private RabbitTemplate rabbitTemplate;

    private void send() {
        //模拟某人在商店买**中奖了
        rabbitTemplate.convertAndSend("news-exchange","province.city.street.shop","有人中了大奖");
    }

执行结果

来自省TV的消息:有人中**了
来自市TV的消息:有人中**了
来自街区TV的消息:有人中**了
Headers 头交换机
两个消费者,一个全匹配,一个任意匹配

@Component
public class MyHeadListener {


    /**
     * 任意匹配
     *
     * @param msg
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("headQueue-one"),
            exchange = @Exchange(value = "myHeadExchange", type = ExchangeTypes.HEADERS),
            arguments = {
                    @Argument(name = "key-one", value = "1"),
                    @Argument(name = "key-two", value = "2"),
                    @Argument(name = "x-match", value = "any")
            }))
    public void anyMatchOnMessage(String msg) {
        System.out.println("来自 headQueue-one " + msg);
    }


    /**
     * 全匹配
     *
     * @param msg
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("headQueue-two"),
            exchange = @Exchange(value = "myHeadExchange", type = ExchangeTypes.HEADERS),
            arguments = {
                    @Argument(name = "key-one", value = "1"),
                    @Argument(name = "x-match", value = "all")
            }))
    public void allMatchOnMessage(String msg) {
        System.out.println("来自 headQueue-two " + msg);
    }

}

生产者,上面的两个消费者都能收到消息

    @Autowired
    private RabbitTemplate rabbitTemplate;

    private void send() {
        rabbitTemplate.convertAndSend("myHeadExchange", "", "this is a message", message -> {
            MessageProperties properties = message.getMessageProperties();
            properties.setHeader("key-one", "1");
            return message;
        });
    }

关于@RabbitListener
先看另一种申明消费者的写法

   static final String topicExchangeName = "spring-boot-exchange";

    static final String queueName = "spring-boot";

    /**申明队列
     * @return 
     */
    @Bean
    Queue queue() {
        return new Queue(queueName, false);
    }

    /**申明交换机
     * @return 
     */
    @Bean
    TopicExchange exchange() {
        return new TopicExchange(topicExchangeName);
    }

    /**将交换机与队列绑定
     * @param queue 
     * @param exchange
     * @return
     */
    @Bean
    Binding binding(Queue queue, TopicExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with("foo.bar.#");
    }


    /** 申明处理消息的适配器,指明用哪个方法处理接收消息,
     * 这里指明了用Receiver的receiveMessage()方法接收消息
     * @param receiver 这是自己注入的对象,里面有一个receiveMessage方法
     * @return
     */
    @Bean
    MessageListenerAdapter listenerAdapter(Receiver receiver) {
        return new MessageListenerAdapter(receiver, "receiveMessage");
    }

    /** 申明一个Listener容器,其实就是对应一个消费者(不知道我这样理解对不对)
     * 需要设置连接消息,指明监听哪个queue,受到消息的处理方法
     * @param connectionFactory 
     * @param listenerAdapter
     * @return
     */
    @Bean
    SimpleMessageListenerContainer container(ConnectionFactory connectionFactory,
                                             MessageListenerAdapter listenerAdapter) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames(queueName);
        container.setMessageListener(listenerAdapter);
        return container;
    }

上面是springBoot的示例写法,可以看到,要申明队列或是交换机或是绑定它们,只需要构造对应的对象放到spring容器里就行,,而申明一个监听者(消费者)则需要构造一个MessageListenerContainer 对象,ListenerContainer至少需要知道如何连接到代理(ConnectionFactory),监听哪个队列的消息(QueueNames)以及收到消息的处理方法(MessageListenerAdapter).

到这里我们可以猜到,一个@RabbitListener就对应一个MessageListenerContainer,ConnectionFactory被自动注入,队列名称可以用queues,queuesToDeclare,或是bindings里面的queue指定,收到消息的处理方法用@RabbitHandler指定,或是直接把@RabbitListener打在对应的方法上就行.

如何设置交换机,队列的其它属性如Durable,Auto-delete等: 用@Bean的方式,构建对应对象,设置对应属性就行,用注解@Queue或是@Exchange里面都有对应属性,设置就行

如何设置消费者的其它属性,如消息确认方式,预处理消息数量等: 用@Bean的方式,构建MessageListenerContainer的对象,设置对应属性就行,用注解的话@RabbitListener里指定containerFactory,然后在containerFactory里设置