-
Notifications
You must be signed in to change notification settings - Fork 64
Expand file tree
/
Copy pathRabbitMQ-DirectExchange.java
More file actions
134 lines (125 loc) · 4.39 KB
/
Copy pathRabbitMQ-DirectExchange.java
File metadata and controls
134 lines (125 loc) · 4.39 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
----------------------------
RabbitMQ-Routing |
----------------------------
# 传说中的路由模式
- error ->
|- info -> [][][] C1
P -> X
|- error ->
|- info -> [][][] Cn
- warn ->
p :消息生产者
X :交换机(类型是:direct(路由))
C1 :消费者1
C2 :消费者N
* 生产者把消息发送给交换机(路由类型)
* 消费者创建N个队列,指定一个'路由key',绑定到交换机
* 生产者产生消息的时候,就需要指定'路由key'
* 消息到达交换机的时候,就会根据不同的'路由key',把消息推送到不同的消费者队列.可以推送到N个'相同路由key'的队列
* 这是一种有选择性的消息接收
* 传说中的'路由key',其实就是个字符串... ...
/**
工具类
*/
public static Connection getConnection() throws Exception {
//定义连接工厂
ConnectionFactory factory = new ConnectionFactory();
//设置服务地址
factory.setHost("localhost");
//端口
factory.setPort(5672);
//设置账号信息，用户名、密码、vhost
factory.setVirtualHost("/kevinblandy");
factory.setUsername("kevin");
factory.setPassword("a12551255");
// 通过工程获取连接
Connection connection = factory.newConnection();
return connection;
}
/**
生产者
*/
private final static String EXCHANGE_NAME = "test_exchange_direct"; //交换机名称
public static void main(String[] argv) throws Exception {
// 获取到连接以及mq通道
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
// 声明exchange,类型是 direct
channel.exchangeDeclare(EXCHANGE_NAME, "direct");
// 消息内容
String message = "Hello World!";
/**
发送消息到路由.并且指定 路由key = "key2"
*/
channel.basicPublish(EXCHANGE_NAME, "key2", null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");
channel.close();
connection.close();
}
/**
消费者1
*/
private final static String QUEUE_NAME = "test_queue_work_direct1";
private final static String EXCHANGE_NAME = "test_exchange_direct";
public static void main(String[] argv) throws Exception {
// 获取到连接以及mq通道
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
// 声明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
/*
绑定队列到交换机
并且指定了一个路由key = "update"
*/
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
/*
再绑定一个队列
并且指定了一个路由key = "delete"
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");
*/
// 同一时刻服务器只会发一条消息给消费者
channel.basicQos(1);
// 定义队列的消费者
QueueingConsumer consumer = new QueueingConsumer(channel);
// 监听队列，手动返回完成
channel.basicConsume(QUEUE_NAME, false, consumer);
// 获取消息
while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
System.out.println(" [x] Received '" + message + "'");
Thread.sleep(10);
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
}
/**
消费者N
*/
private final static String QUEUE_NAME = "test_queue_work_direct2";
private final static String EXCHANGE_NAME = "test_exchange_direct";
public static void main(String[] argv) throws Exception {
// 获取到连接以及mq通道
Connection connection = ConnectionUtil.getConnection();
Channel channel = connection.createChannel();
// 声明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
/*
绑定队列到交换机
并且指定了一个路由key = "key2"
*/
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key2");
// 同一时刻服务器只会发一条消息给消费者
channel.basicQos(1);
// 定义队列的消费者
QueueingConsumer consumer = new QueueingConsumer(channel);
// 监听队列，手动返回完成
channel.basicConsume(QUEUE_NAME, false, consumer);
// 获取消息
while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
System.out.println(" [x] Received '" + message + "'");
Thread.sleep(10);
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
}