前言
AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦。
业务需求
后端（集群）通过websocket往各自维持的websocket session推送消息，如果采用每个实例监听同一个queue，那么生产者往该queue中推送一条消息，该消息只能被集群中某个实例消费一次。
想要实现后端每个实例同时消费该消息，便可采用RabbitMQ中的topic模式，即每个实例启动时，新建一个topic类型的exchange，routing key为"#.queue"，并且每个实例的queue与该exchange绑定。每个实例中，根据hostname+".queue"（queue为配置文件中指定的默认队列名称）来创建各自的queue，这样每个实例都各自监听自己的queue。
生产者往该exchange中推送消息，routing key使用默认队列（或者每台实例自己的队列都可以，只要能匹配到"#.queue"即可），这样exchange接收到消息后，会根据routingkey来匹配与该exchange绑定的queues，并将消息发送到符合条件的queues中，这样每台实例都能收到该消息并消费。
RabbitMQ介绍
RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。
整体结构图

概念介绍
- Broker：简单来说就是消息队列服务器实体
- Virtual Host：虚拟主机，一个Broker里可以开设多个virtual host，用作不同用户的权限分离
- Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列
- Queue：消息队列载体，每个消息都会被投入到一个或多个队列
- Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来
- Routing Key：路由关键字，exchange根据这个关键字进行消息投递
- Publisher：消息生产者
- Consumer：消息消费者
- Channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务
Exchange详解
Exchange意思为交换机，从图中可以看出，publisher发送消息，先进入Exchange，然后由Exchange分配到队列Queue。
1. Direct-Exchange Direct Exchange是RabbitMQ Broker的默认Exchange，在此类型下，不必指定routing key的名字，创建的Queue有一个默认的routing key，一般与创建的Queue同名。
2. Topic-Exchange
Topic Exchange是根据routing key和Exchange的类型将message发送到一个或者多个Queue中，可以通过它来实现pub/sub模式，即发布订阅。
注：“#”表示0个或若干个关键字，“*”表示一个关键字
如果Exchange没有发现能够与RouteKey匹配的Queue，则会抛弃此消息
3. Fanout-Exchange
此类型的Exchange比较特殊，会忽略routing key的存在，直接将message广播到所有的Queue中。
4. Headers-Exchange
Headers Exchange不同于上面三种Exchange，它是根据Message的一些头部信息来分发过滤Message，忽略routing key的属性，如果Header信息和message消息的头信息相匹配，那么这条消息就匹配上了。
Spring集成RabbitMQ实现Topic模式
1. rabbitmq.properties（mq配置文件）
rabbitmq.host=127.0.0.1
rabbitmq.port=5672
rabbitmq.username=admin
rabbitmq.password=admin
rabbitmq.my.exchange=my-exchange
rabbitmq.my.queue=my-queue
2. 自动注入配置类（注意：需要创建自定义rabbitAdmin，否则启动时不会自动创建exchange、queue、bindings相关信息，具体可以查看RabbitAdmin类中的initialize()方法）
/**
* @Description: rabbit mq相关配置
* @author yehaixiao
* @date 2018年5月30日
*/
@Configuration
public class RabbitMQConfig { private static final Logger LOGGER = Logger.getLogger(RabbitMQConfig.class); @Value("${rabbitmq.host}")
String host;
@Value("${rabbitmq.port}")
int port;
@Value("${rabbitmq.username}")
String username;
@Value("${rabbitmq.password}")
String password;
@Value("${rabbitmq.my.exchange}")
String exchange;
@Value("${rabbitmq.my.queue}")
String queue; /**
* 创建rabbit mq连接工厂
*/
@Bean(name = "rabbitConnectionFactory")
public ConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setAddresses(host + ":" + port);
connectionFactory.setUsername(username);
connectionFactory.setPassword(password);
connectionFactory.setPublisherConfirms(true);
return connectionFactory;
} /**
* 自定义connectionFactory，需要声明rabbitAdmin 内部initialize()方法会进行exchanges、queues、bindings声明
*/
@Bean
public RabbitAdmin rabbitAdmin() {
return new RabbitAdmin(connectionFactory());
} /**
* 创建amqp消息模版，用于发送者发送消息
*/
@Bean
public AmqpTemplate amqpTemplate() {
RabbitTemplate amqpTemplate = new RabbitTemplate(connectionFactory());
// json消息转化
amqpTemplate.setMessageConverter(messageConverter());
return amqpTemplate;
} /**
* 创建topic类型交换器
*/
@Bean
public TopicExchange topicExchange() {
return new TopicExchange(exchange, false, true);
} /**
* 根据hostname创建queue（目的：为了实现多个集群中多个实例共享一条消息）
* durable:是否持久化
* exclusive:仅创建者可以使用的私有队列，断开后自动删除
* auto-delete:当所有消费端连接断开后，是否自动删除队列
*/
@Bean
public Queue queue() {
InetAddress netAddress = null;
try {
netAddress = InetAddress.getLocalHost();
} catch (UnknownHostException e) {
e.printStackTrace();
}
String hostName = netAddress.getHostName();
String queueName = String.format("%s." + queue, hostName);
LOGGER.info("dynamic create queue name：" + queueName);
Queue queue = new Queue(queueName, false, false, true);
return queue;
} /**
* 将队列queue与exchange绑定，binding_key为#.queue，模糊匹配
*/
@Bean
public Binding binding(Queue queue, TopicExchange topicExchange) {
return BindingBuilder.bind(queue).to(topicExchange).with("#." + queue);
} /**
* 创建消息监听器
*/
@Bean
public MessageListener customizeMessageListener() {
return new CustomizeMessageListener();
} /**
* 创建json消息转化器
*/
@Bean
public MessageConverter messageConverter() {
Jackson2JsonMessageConverter fastJsonMessageConverter = new Jackson2JsonMessageConverter();
return fastJsonMessageConverter;
} /**
* 绑定queue和listener关系
*/
@Bean
public SimpleMessageListenerContainer mqMessageContainer(Queue queue, MessageListener customizeMessageListener) {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
container.setQueues(queue);
container.setExposeListenerChannel(true);
container.setAcknowledgeMode(AcknowledgeMode.AUTO);
container.setMessageListener(customizeMessageListener);
return container;
}
}
3. 消息消费者（通过实现MessageListener来重写onMessage()方法）
/**
* @Description: 消息监听器
* @author yehaixiao
* @date 2018年5月24日
*/
@Service
public class CustomizeMessageListener implements MessageListener { private static final Logger LOGGER = LoggerFactory.getLogger(CustomizeMessageListener.class); @Override
public void onMessage(Message message) {
try {
String queue = message.getMessageProperties().getConsumerQueue();
String msg = new String(message.getBody());
LOGGER.info("consumer msg : {}, from queue : {}", msg, queue);
} catch (Exception e) {
LOGGER.error("consumer message error！", e);
}
}
}
4. 消息生产者（通过AmqpTemplate发送消息）
/**
* @Description: 消息生产者
* @author yehaixiao
* @date 2018年5月28日
*/
@Service
public class MessageProducer { private static final Logger LOGGER = LoggerFactory.getLogger(MessageProducer.class); @Value("${rabbitmq.my.exchange}")
private String exchange; @Value("${rabbitmq.my.queue}")
private String queue; @Resource
private AmqpTemplate amqpTemplate; public void sendMessage(Object message) {
LOGGER.info("send message : {} , to queue : {}", JSON.toJSONString(message), queue);
// message是需要传递的信息, 指定特定的queue
amqpTemplate.convertAndSend(exchange, queue, message);
}
}