0
点赞
收藏
分享

微信扫一扫

RabbitMQ-路由和主题模式

概述

Topic 类型的Exchange与Direct相比,都是可以根据 RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符!这种模式Routingkey 一般都是由一个或多个单词组成,多个单词之间以 ”.” 分割,例如: item.insert

RabbitMQ-路由和主题模式_消息发送

  • # 统配符
  • * (star) can substitute for exactly one word:匹配不多不少恰好1个词
  • # (hash) can substitute for zero or more words:匹配0个或多个词

# 示例如下所示:

audit.#:匹配 audit.irs.corporate 或者 audit.irs 等

audit.*:只能匹配 audit.irs

创建生产者

/**
* @author: BNTang
*/
public class Producer {

@Test
public void sendMessage() throws Exception {
Connection connection = RabbitMQUtil.getConnection();

Channel channel = connection.createChannel();

String exchangeName = "topic";

channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC);

// 声明一个路由 key
String routingKey = "user";

channel.basicPublish(exchangeName, routingKey, null, ("我是一个topic类型的交换机消息 + routingKey = " + routingKey).getBytes());

System.out.println("消息发送成功");

RabbitMQUtil.closeChannelAndConnection(channel, connection);
}
}

创建消费者 1

/**
* @author BNTang
*/
public class Consumer1 {

@Test
public void receiveMessage() throws Exception {

Connection connection = RabbitMQUtil.getConnection();

Channel channel = connection.createChannel();

String exchangeName = "topic";

channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC);

// 得到一个临时队列
String queue = channel.queueDeclare().getQueue();

// 绑定队列到交换机
channel.queueBind(queue, exchangeName, "user.*");

// 消费消息
channel.basicConsume(queue, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者【1】收到消息" + new String(body));
}
});

System.in.read();
}
}

创建消费者 2

/**
* @author BNTang
*/
public class Consumer2 {

@Test
public void receiveMessage() throws Exception {
Connection connection = RabbitMQUtil.getConnection();

Channel channel = connection.createChannel();

String exchangeName = "topic";

channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC);

// 得到一个临时队列
String queue = channel.queueDeclare().getQueue();

// 绑定队列到交换机
channel.queueBind(queue, exchangeName, "user.#");

// 消费消息
channel.basicConsume(queue, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者【2】收到消息" + new String(body));
}
});

System.in.read();
}
}

测试

先启动消费者1和消费者2,再分别发送routingKey=user和user.add和user.add.insert等类型的消息。查看结果即可。

举报

相关推荐

0 条评论