RabbitMQ：四种ExChange用法
沙落雁
收藏于2017-09-19
转藏1次
RabbitMQ发送消息时，都是先把消息发送给ExChange（交换机），然后再分发给有相应RoutingKey（路由）关系的Queue（队列）。
ExChange和Queue之前是多对多的关系。
RabbitMQ 3.0之后创建ExChange时，有四种类型可选“fanout、direct、topic、headers”。
一、fanout
当向一个fanout发送一个消息时，RoutingKey的设置不起作用。
消息会被发送给同一个交换机下的所有队列，每个队列接收到的消息是一样的；
一个队列内有所有消费者（包含那些并没有相应RoutingKey的
消费者
），将平分队列接收到的消息
。
----------------消息生产者----------------
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(S_RabbitMQ.QUEUE_IP);// MQ主机
factory.setPort(S_RabbitMQ.QUEUE_PORT);// MQ端口
factory.setUsername(S_RabbitMQ.QUEUE_USER);// MQ用户名
factory.setPassword(S_RabbitMQ.QUEUE_PWD);// MQ密码
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 声明路由名字和类型
channel.exchangeDeclare(EXCHANGE_NAME, "fanout", true, false, null);
String message = "hello world! ";
for(int i=0;i<100;i++)
{
channel.basicPublish(EXCHANGE_NAME, "", null, (message+i).getBytes());
}
System.out.println("Sent msg finish");
channel.close();
connection.close();
----------------消息消费者 ----------------
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(S_RabbitMQ.QUEUE_IP);// MQ主机
factory.setPort(S_RabbitMQ.QUEUE_PORT);// MQ端口
factory.setUsername(S_RabbitMQ.QUEUE_USER);// MQ用户名
factory.setPassword(S_RabbitMQ.QUEUE_PWD);// MQ密码
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
//声明路由名字和类型
channel.exchangeDeclare(EXCHANGE_NAME, "fanout", true, false, null);
//
声明
队列
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
//绑定路由和队列
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "routkey2", null);
System.out.println(" Waiting for msg....");
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) {
String message = "";
try
{
message = new String(body, "UTF-8");
}
catch (UnsupportedEncodingException e)
{
e.printStackTrace();
}
catch (Throwable ex)
{
ex.printStackTrace();
}
System.out.println("Received msg='" + message + "'");
}
};
channel.basicConsume(QUEUE_NAME, true, consumer);
二、direct
当向一个direct发送一个消息时，
消息会被发送给同一个交换机下的
拥有相应RoutingKey的队列，
每个队列接收到的消息是一样的；
一个队列内拥有相应RoutingKey的消费者，将平分队列接收到的消息。
----------------消息生产者
----------------
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(S_RabbitMQ.QUEUE_IP);// MQ主机
factory.setPort(S_RabbitMQ.QUEUE_PORT);// MQ端口
factory.setUsername(S_RabbitMQ.QUEUE_USER);// MQ用户名
factory.setPassword(S_RabbitMQ.QUEUE_PWD);// MQ密码
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 声明路由名字和类型
channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, null);
String message = "hello world! ";
for(int i=0;i<100;i++)
{
channel.basicPublish(EXCHANGE_NAME, "routingkey1", null, (message+i).getBytes());
}
System.out.println("Sent msg is '" + message + "'");
channel.close();
connection.close();
----------------消息消费者 ----------------
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(S_RabbitMQ.QUEUE_IP);// MQ主机
factory.setPort(S_RabbitMQ.QUEUE_PORT);// MQ端口
factory.setUsername(S_RabbitMQ.QUEUE_USER);// MQ用户名
factory.setPassword(S_RabbitMQ.QUEUE_PWD);// MQ密码
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
//声明路由名字和类型
channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, null);
//声明队列
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
//绑定路由和队列
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "routingkey1", null);
System.out.println(" Waiting for msg....");
Consumer consumer = new DefaultConsumer(channel)
{
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
byte[] body)
{
String message = "";
try
{
message = new String(body, "UTF-8");
}
catch (UnsupportedEncodingException e)
{
e.printStackTrace();
}
catch (Throwable ex)
{
ex.printStackTrace();
}
System.out.println("1 Received msg='" + message + "'");
}
};
channel.basicConsume(QUEUE_NAME, true, consumer);
三、topic
当向一个topic发送一个消息时
，
消息会被发送给同一个交换机下的
拥有相应RoutingKey的队列，
每个队列接收到的消息是一样的；
一个队列内有所有消费者（包含那些并没有相应RoutingKey的
消费者
），将平分队列接收到的消息
。
----------------消息生产者
----------------
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(S_RabbitMQ.QUEUE_IP);// MQ主机
factory.setPort(S_RabbitMQ.QUEUE_PORT);// MQ端口
factory.setUsername(S_RabbitMQ.QUEUE_USER);// MQ用户名
factory.setPassword(S_RabbitMQ.QUEUE_PWD);// MQ密码
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 声明路由名字和类型
channel.exchangeDeclare(EXCHANGE_NAME, "topic", true, false, null);
String message = "hello world! ";
// int i=101;
for (int i = 0; i < 100; i++)
{
channel.basicPublish(EXCHANGE_NAME, "routingkey1", null, (message + i).getBytes());
}
System.out.println("Sent msg is '" + message + "'");
channel.close();
connection.close();
----------------消息消费者
----------------
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(S_RabbitMQ.QUEUE_IP);// MQ主机
factory.setPort(S_RabbitMQ.QUEUE_PORT);// MQ端口
factory.setUsername(S_RabbitMQ.QUEUE_USER);// MQ用户名
factory.setPassword(S_RabbitMQ.QUEUE_PWD);// MQ密码
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 声明路由名字和类型
channel.exchangeDeclare(EXCHANGE_NAME, "topic", true, false, null);
//声明队列
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
//绑定路由和队列// 把队列绑定到路由上并指定headers
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "routingkey1", null);
System.out.println("1 Waiting for msg....");
Consumer consumer = new DefaultConsumer(channel)
{
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
byte[] body)
{
String message = "";
try
{
message = new String(body, "UTF-8");
}
catch (UnsupportedEncodingException e)
{
e.printStackTrace();
}
catch (Throwable ex)
{
ex.printStackTrace();
}
System.out.println("1 Received msg='" + message + "'");
}
};
channel.basicConsume(QUEUE_NAME, true, consumer);
特别重申：本篇文档资料为 “好网角收藏夹” 注册用户(收藏家)上传共享，仅供参考之用，请谨慎辨别，不代表本站任何观点。
好网角收藏夹为网友提供资料整理云存储服务，仅提供信息存储共享平台。
信息删除举报或发邮件到：dongye2016@qq.com
好网角收藏夹为网友提供资料整理云存储服务，仅提供信息存储共享平台。
信息删除举报或发邮件到：dongye2016@qq.com
类似文章
消息中间件 MQ 企业级方案设计
king108 2013-06-07
消息队列中间件mq的使用场景
藏家004 2018-10-31
Kafka，Mq，Redis作为消息队列使用时的差异？
凌云 2018-08-02
java多线程实现mq消息处理超时监控
泡泡 2017-05-09
MQ选型对比RabbitMQ RocketMQ ActiveMQ Kafka
凌云 2018-07-18
用Spring cloud Stream来开发基于MQ消息驱动的微服务 - 邓志国 - 博客园
凌云 2019-04-09
RabbitMQ高级消息队列
凌云 2019-03-26
rabbitmq 消息确认机制
凌云 2019-02-13
RabbitMQ消息应答------ack机制
凌云 2019-02-13
RabbitMQ 消息确认以及消息消费方处理消息时候抛出了异常以
凌云 2019-03-26
最新转藏文章
《大乘离文字普光明藏经》注音精美版
藏家569 2025-03-28
财税2016年36号文件【全文】
藏家838 2025-03-29
最新106剂内部验方！只发一次，一定要看看！
奇美拉 2025-03-29
他贤明，我愚笨，何须争斗。
许老头 2025-03-28
于宏洁 利未记 第七讲
藏家942 2025-03-28
《声律启蒙》全文注音版带拼音
mei 2025-03-28
最新106剂内部验方！只发一次，一定要看看！
藏家113 2025-03-28
《增广贤文》全文
藏家518 2025-03-28
浊世如意宝珠：不空罥索神咒（加句版）及音频
藏家372 2025-03-28
大悲观音法系：摄持诸大天的不空绢索心咒之秘释与修持
藏家372 2025-03-28
TOP