今天小编说的是前两天在工作中遇到的问题:
在项目中使用rabbitmq 发送消息和接收消息,如果项目在一台机器上部署则使用Direct方式即可,但是如果把项目部署到n台机器上,发送一条消息,则n台机器都能同时接收消息,则需要使用Fanout方式来实现,而且需要创建n个队列进行接收消息.这样就实现了一条消息被多个消费者同时消费。
首先,配置FanoutExchange
为一个bean
@Bean
FanoutExchange testFanoutExchange() {
FanoutExchange testFanoutExchange = newFanoutExchange(定义的的Exchange名);
rabbitAdmin.declareExchange(testFanoutExchange);
return testFanoutExchange ;
}
其次,定义一个队列: AnonymousQueue
这个类是代表一个匿名的、不耐用的、独占的、自动删除的队列。使用这个类会随机生成一个队列名,项目启动则随机生成,停止则自动删除,是一个短暂的队列
@Bean
Queue queueTest() {
Queue queue = new AnonymousQueue();
return queue;
}
最后,将队列绑定到Exchange上
@Bean
Binding bindingExchangeTest(Queue queueTest, FanoutExchange testFanoutExchange) {
Binding binding = BindingBuilder.bind(queueTest).to(testFanoutExchange);
return binding;
}
那么如何监听消息队列呢?
我们可以通过@RabbitListener标签进行监听,通过队列名进行识别
@RabbitListener(queues = "#{queueTest.name}")
@RabbitHandler
public void sendBizApiUpdateToBackend(Message message) {
List<SysBizApiModel> sysBizApiModelList = new ArrayList<>();
try {
sysBizApiModelList = objectMapper.readValue(message.getBody(), new TypeReference<List<SysBizApiModel>>(){});
} catch (IOException e) {
e.printStackTrace();
LOGGER.error("消息反序列化失败, message = {}", message, e);
}
......
//TODO
}