- 在发布订阅模式中,消息不会被直接发送给消费者队列,消息发送至exchange交换机,在使用的时候先声明交换机,发布消息的时候也是发布到exchange,但是没有绑定队列的消息会被丢弃,因为exchange无法存储消息
@RequestMapping("psMode")
public ResultMsgInfo psMode(String message){
try{
Connection connection = RabbitConnectUtils.getRabbitConnection();
Channel channel = connection.createChannel();
//声明队列
channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
for (int i = 0; i < 101 ; i++) {
String msg = message+i;
channel.basicPublish(EXCHANGE_NAME,"",null,msg.getBytes());
logger.info("发送消息到队列:{},消息内容:{}.",HELLO_QUEUE,msg);
}
channel.close();
connection.close();
}catch (Exception e){
e.printStackTrace();
return ResultMsgInfo.error();
}
return ResultMsgInfo.success();
}
- 客户端代码的主要流程是声明队列,将队列绑定到交换机上:
public void getRabbitMessage3() throws Exception{
// 获取到连接以及mq通道
Connection connection = RabbitConnectUtils.getRabbitConnection();
Channel channel = connection.createChannel();
// 声明队列
channel.queueDeclare(FANOUT_QUEUE, false, false, false, null);
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
// 绑定队列到交换机
channel.queueBind(FANOUT_QUEUE, EXCHANGE_NAME, "");
// 同一时刻服务器只会发一条消息给消费者
// 定义队列的消费者
// 监听队列,手动返回完成
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println(" [x] Received '" + message + "'");
}
};
channel.basicConsume(FANOUT_QUEUE, true, consumer);
}
direct模式实际上是可以理解为在发布订阅模式的基础上填充分类,在声明exchange和队列绑定exchange的时候指定第三个参数,即directKey,消息会被投放至相应的队列,相当于多了一级分类,另外发布订阅模式中消息会被发送至所有的消费者,但是只有一个消费者能够获取到消息