SpringBoot整合RabbitMQ之广播模式

广播模式的简单描述

通过对RabbitMQ的学习，可以发现，消息都是通过交换器发送至队列的，一条消息只能被一个消费者处理，实际开发中还会有一种情况，就是一条消息需要被多个消费者处理，就是广播的形式；广播的模式需要使用到FanoutExchange（散列交换器），FanoutExchange 会将消息发送至每一个与之绑定的队列中。

具体使用： 1. 引入Rabbit MQ依赖



	org.springframework.boot
	spring-boot-starter-amqp

2. 修改配置文件

spring: 
  rabbitmq:
  	# [主机IP，不加默认为本机地址]
    host: 127.0.0.1
    # [端口号，不加默认为5672]
    port: 5672
    #消费端配置
    listener:
      simple:
        #自动签收auto  手动 manual
        acknowledge-mode: manual

3. 编写配置类

配置广播模式的消息队列：
创建一个名为RabbitmqConfigFanout的配置类，在该配置类上使用@Configuration 注解标明这是一个配置类；
通过@Bean注解使用注解的name属性来指定消息队列名，创建两个消息队列，其返回类型为Queue；
同样地，通过@Bean注解创建一个广播模式的交换机，其返回类型为FanoutExchange。

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@Configuration
public class RabbitmqConfigFanout {
	
    private final Logger log = LoggerFactory.getLogger(RabbitmqConfigFanout.class);

    
    
    @Bean(name = MqConstants.QUEUE_A)
    public Queue queueA() {
        Queue queue = new Queue(MqConstants.QUEUE_A,true);
        log.info("队列：[{}]初始化成功................",queue.getName());
        return queue;
    }
    
    
    @Bean(name = MqConstants.QUEUE_B)
    public Queue queueB() {
        Queue queue = new Queue(MqConstants.QUEUE_B,true);
        log.info("队列：[{}]初始化成功................",queue.getName());
        return queue;
    }

    
    @Bean
    public FanoutExchange fanoutExchange() {
        FanoutExchange fanoutExchange = new FanoutExchange(MqConstants.EXCHANGE);
        log.info("交换机：[{}]初始化成功................",fanoutExchange.getName());
        return fanoutExchange;
    }

    
    
    @Bean
    public Binding bindingQueueA(@Qualifier(MqConstants.QUEUE_A) Queue queue, FanoutExchange fanoutExchange) {
        Binding binding = BindingBuilder.bind(queue).to(fanoutExchange);
        log.info("消息队列：[{}] 成功绑定到交换机：[{}] ................",queue.getName(),fanoutExchange.getName());
        return binding;
    }
    
    @Bean
    public Binding bindingQueueB(@Qualifier(MqConstants.QUEUE_B) Queue queue, FanoutExchange fanoutExchange) {
        Binding binding = BindingBuilder.bind(queue).to(fanoutExchange);
        log.info("消息队列：[{}] 成功绑定到交换机：[{}] ................",queue.getName(),fanoutExchange.getName());
        return binding;
    }
}

4. 编写生产者（发送者）类

创建一个发送者类，在类上使用组件注解@Component表明这是一个组件，如果不添加此注解，在使用@Autowired自动注入（自动装配）时，启动项目会报错误。
在类的内部通过@Autowired自动注入RabbitMQ的消息队列的模板（RabbitTemplate）。
编写一个发送方法，使用模板RabbitTemplate的convertAndSend()方法发送消息。

具体代码：

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

@Component
public class FanoutSender {
	
    private final Logger log = LoggerFactory.getLogger(FanoutSender.class);
    
	@Autowired
    private RabbitTemplate rabbitTemplate;
	
	public void send() {
		 try {
            String str = "Fanout";
			System.out.println("广播模式发送者Sender:" + str);
        } finally {
            rabbitTemplate.convertAndSend(MqConstants.EXCHANGE, "", "要发送的消息");
        }
	}
}

5. 编写消费者（接收者）类

创建一个接收者类，类上使用注解@Component和RabbitMQ的监听注解@RabbitListener，其中@RabbitListener使用其属性queues绑定需要监听的队列。
创建一个接收方法process()，方法上使用处理注解@RabbitHandler。

接收者A的具体代码如下：

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = {MqConstants.QUEUE_A})
public class FanoutReceiverA {

	
	@RabbitHandler
	public void process(String msg, Message message, Channel channel) {
		long startTime = System.currentTimeMillis();
        log.info("消息队列：[{}]，接收到消息：[{}]",MqConstants.QUEUE_A, msg);

        try {
            //处理消息
            //此处是对消息的处理

            //签收消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
            log.info("消息队列[{}],接收到消息：[{}],消费完成 耗时：[{}ms]",MqConstants.QUEUE_A, msg, System.currentTimeMillis() - startTime);
        }catch (Exception e) {
            try {
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                log.info("消息队列[{}],接收到消息：[{}],消费失败 耗时：[{}ms]",MqConstants.QUEUE_A, msg, System.currentTimeMillis() - startTime);
            }catch (IOException ioException) {
                throw new RuntimeException(ioException);
            }
        }
	}
}

接收者B的具体代码如下：

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
@RabbitListener(queues = {MqConstants.QUEUE_B})
public class FanoutReceiverB {

	
	@RabbitHandler
	public void process(String msg, Message message, Channel channel) {
		long startTime = System.currentTimeMillis();
        log.info("消息队列：[{}]，接收到消息：[{}]",MqConstants.QUEUE_B, msg);

        try {
            //处理消息
            //此处是对消息的处理

            //签收消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
            log.info("消息队列[{}],接收到消息：[{}],消费完成 耗时：[{}ms]",MqConstants.QUEUE_B, msg, System.currentTimeMillis() - startTime);
        }catch (Exception e) {
            try {
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                log.info("消息队列[{}],接收到消息：[{}],消费失败 耗时：[{}ms]",MqConstants.QUEUE_B, msg, System.currentTimeMillis() - startTime);
            }catch (IOException ioException) {
                throw new RuntimeException(ioException);
            }
        }
	}
}

6. 进行测试

编写一个测试类，将生产者的类使用@Autowired 注解进行注入，调用生产方法，最终达到效果

SpringBoot整合RabbitMQ之广播模式

Java相关栏目本月热门文章