版权声明:本文为博主原创文章,未经博主允许不得转载。 https://blog.csdn.net/u010739551/article/details/88417313
概述
消息发送到topic类型的exchange上时不能随意指定routing_key(一定是指由一系列由点号连接单词的字符串,单词可以是任意的,但一般都会与消息或多或少的有些关联)。Routing key的长度不能超过255个字节。
Binding key也一定要是同样的方式。Topic类型的exchange就像一个直接的交换:一个由生产者指定了确定routing key的消息将会被推送给所有Binding key能与之匹配的消费者。然而这种绑定有两种特殊的情况:
- *(星号):可以(只能)匹配一个单词
- #(井号):可以匹配多个单词(或者零个)
例子
生产者
package com.luke.sbdubbo.user.mq.topic;
import com.rabbitmq.client.*;
public class Producer {
public static final String EXCHANGE_NAME = "exchange_topic_demo";
public static final String QUEUE_NAME1 = "queue_topic_1";
public static final String QUEUE_NAME2 = "queue_topic_2";
public static final String IP_ADDRESS = "192.168.0.114";
public static final int PORT = 5672;
public static void main(String[] args) throws Exception{
//创建连接工厂
Address[] addresses = new Address[]{
new Address(IP_ADDRESS,PORT)
};
ConnectionFactory factory = new ConnectionFactory();
factory.setUsername("root");
factory.setPassword("root");
//获取连接通道
Connection connection = factory.newConnection(addresses);
Channel channel = connection.createChannel();
//声明交换器和队列,并生成(实际生成环境,建议事先建好交换器、队列,还有绑定关系)
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC,true);
channel.queueDeclare(QUEUE_NAME1,true,false,false,null);
channel.queueDeclare(QUEUE_NAME2,true,false,false,null);
String bindingKey1 = "*.orange.*";
String bindingKey2 = "*.*.rabbit";
String bindingKey3 = "lazy.#";
//队列绑定交换器
channel.queueBind(QUEUE_NAME1,EXCHANGE_NAME,bindingKey1);
channel.queueBind(QUEUE_NAME2,EXCHANGE_NAME,bindingKey2);
channel.queueBind(QUEUE_NAME2,EXCHANGE_NAME,bindingKey3);
//路由键
//String ROUTING_KEY = "aa.orange.bb";
//String ROUTING_KEY = "aa.orange.rabbit";
String ROUTING_KEY = "aa.bbb.rabbit";
//发送消息
for (int i = 1; i <= 10; i++) {
String message = "rabbitmq topic message"+i;
channel.basicPublish(EXCHANGE_NAME,ROUTING_KEY,
MessageProperties.PERSISTENT_TEXT_PLAIN,
message.getBytes("UTF-8"));
}
//关闭连接
channel.close();
connection.close();
}
}
消费者1
package com.luke.sbdubbo.user.mq.topic;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
public class Queue1Consumer {
public static final String QUEUE_NAME1 = "queue_topic_1";
public static final String IP_ADDRESS = "192.168.0.114";
public static final int PORT = 5672;
public static void main(String[] args) throws Exception{
//创建连接工厂
Address[] addresses = new Address[]{
new Address(IP_ADDRESS,PORT)
};
ConnectionFactory factory = new ConnectionFactory();
factory.setUsername("root");
factory.setPassword("root");
//获取连接通道
Connection connection = factory.newConnection(addresses);
Channel channel = connection.createChannel();
channel.basicQos(64);
//消费
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
try {
System.out.println("recv1 message: "+new String(body));
TimeUnit.SECONDS.sleep(1);
//客户端消息确认
channel.basicAck(envelope.getDeliveryTag(),false);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
channel.basicConsume(QUEUE_NAME1,false,consumer);
//关闭连接
/* TimeUnit.SECONDS.sleep(5);
channel.close();
connection.close();*/
}
}
消费者2
package com.luke.sbdubbo.user.mq.topic;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
public class Queue2Consumer {
public static final String QUEUE_NAME2 = "queue_topic_2";
public static final String IP_ADDRESS = "192.168.0.114";
public static final int PORT = 5672;
public static void main(String[] args) throws Exception{
//创建连接工厂
Address[] addresses = new Address[]{
new Address(IP_ADDRESS,PORT)
};
ConnectionFactory factory = new ConnectionFactory();
factory.setUsername("root");
factory.setPassword("root");
//获取连接通道
Connection connection = factory.newConnection(addresses);
Channel channel = connection.createChannel();
channel.basicQos(64);
//消费
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
try {
System.out.println("recv2 message: "+new String(body));
TimeUnit.SECONDS.sleep(1);
//客户端消息确认
channel.basicAck(envelope.getDeliveryTag(),false);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
channel.basicConsume(QUEUE_NAME2,false,consumer);
//关闭连接
/*TimeUnit.SECONDS.sleep(5);
channel.close();
connection.close();*/
}
}
结果