Work Queues
参考demo：G:\学习资料\java\java学习文档\java技术文档\消息队列文档\rabbimitmq\cd-xufei-rabbiitmq-parent\cd-xufei-rabbitmq-01 => work
能够实现流量的消峰：所有队列消费的数据之和 才是原来的所有数据
# 生产者写法
查看示例
public class Producer {
/**
* 申明队列的名字
*/
private static final String QUEUE_NAME = "work_queue";
public static void main(String[] args) throws Exception {
// 获取连接
Connection connection = ConnectionUtils.getConnection();
// 建立通道
Channel channel = connection.createChannel();
// 申明一个队列，如果队列不存在会创建队列
/**
*第一个参数：队列的名字
*第二个参数：是否持久化 :这个数据 发送到队列去的时候 这个数据 是否要持久化到数据库(内存中)中去
*第三个参数：是否排外
* 第一个意思：是否允许这个通道以外的 其他消费者来消费这个数据
* 第二个意思：连接关闭之后 这个队列是否自动删除
*第四个参数：是否自动删除
* 表示的是 当最后一个连接退出的时候是否要删除这个队列
*第五个参数：创建队列 附带的头的信息(TTL队列 死信队列)
*
*/
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 生产者 要向这个队列中 发送一条消息
/**
* 第一个参数：交换机的名字
* 第二个参数：路由的key 因为这里是将数据 直接发送到队列中 所以这里的路由key直接就是 队列的名字
* 第三个参数：发送消息的时候 附带的一些属性信息
* 第四个参数：发送的这个消息体的内容
*/
for (int i = 0; i < 100; i++) {
channel.basicPublish("", QUEUE_NAME, null, ("我是第一个work程序" + i).getBytes());
}
// 这里可以关闭掉这个通道 以及这个连接
// channel.close();
// connection.close();
}
}
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
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
# 消费者1写法
查看示例
public class Consumer1 {
/**
* 申明队列的名字
*/
private static final String QUEUE_NAME = "work_queue";
public static void main(String[] args) throws Exception {
// 第一步获取连接
Connection connection = ConnectionUtils.getConnection();
// 建立通道
Channel channel = connection.createChannel();
// 申明一个队列
/**
*第一个参数：队列的名字
*第二个参数：是否持久化 ：这个数据 发送到队列去的时候 这个数据 是否要持久化到数据库中去
*第三个参数：是否排外
* 第一个意思：是否允许这个通道以外的 其他消费者来消费这个数据
* 第二个意思：连接关闭之后 这个队列是否自动删除
*第四个参数：是否自动删除
* 表示的是 当最后一个连接退出的时候是否要删除这个队列
*第五个参数：创建队列 付带的头的信息(TTL 死信队列)
*
*/
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 下面是设置限流机制
// 告诉他只只要我没有确认 你就不要给我 分配这个任务
/**
* 这里实际上是用在了消费端限流 这样的话
* 第一个参数 表示的是限制 本身消息的大小 如果是 0那么表示的是 不受限制
* 第二个参数
* 表示的意思是 一次性最多 接受的 消息的条目数 比如设为 2 意思这个消费者一次性只取 2 条数据
* 告诉消息队列不要一次性给消费者推送大于N个消息，要推送的前提是，前N个被推送的消息已经手动被确认，已经完成消费之后才能推送下一N个消息
* 第三个参数 表示的是 是否将上面的两个设置应用应用于 整个 通道
* false 表示的是 只是当前这个消费者自己适用
* true 表示将上面的设置应用于整个通道的消费者
*/
// channel.basicQos(1); // 一次性只是接受一个消息 没有确定之前 这里是不会给分配任务的
channel.basicQos(0, 2, false);
// 申明一个消费者
DefaultConsumer consumer = new DefaultConsumer(channel) {
/**
*
* @param consumerTag ： 表示的是 消费者 唯一的标记
* @param envelope：信封 : 消息的一个封装
* @param properties：前面发送消息的时候 写入的键值对
* @param body ：获取到的消息的内容
* @throws IOException
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("第一个队列接受到的消息是" + new String(body));
try {
Thread.sleep(200);
} catch (Exception e) {
e.printStackTrace();
}
/**
* 第一个参数是当前这条消息的标记
* 第二个参数是是否自动应答
* 在这里进行手动确认
*/
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
// 将消费者和队列进行绑定
/**
* 第一个参数：队列的名字
* 第二个参数：是否自动应答
* 第三个参数是 消费者
*/
channel.basicConsume(QUEUE_NAME, false, consumer);
}
}
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
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
# 消费者2写法
查看示例
public class Consumer2 {
/**
* 申明队列的名字
*/
private static final String QUEUE_NAME = "work_queue";
public static void main(String[] args) throws Exception {
// 第一步获取连接
Connection connection = ConnectionUtils.getConnection();
// 建立通道
Channel channel = connection.createChannel();
// 申明一个队列
/**
*第一个参数：队列的名字
*第二个参数：是否持久化 ：这个数据 发送到队列去的时候 这个数据 是否要持久化到数据库中去
*第三个参数：是否排外
* 第一个意思：是否允许这个通道以外的 其他消费者来消费这个数据
* 第二个意思：连接关闭之后 这个队列是否自动删除
*第四个参数：是否自动删除
* 表示的是 当最后一个连接退出的时候是否要删除这个队列
*第五个参数：创建队列 付带的头的信息(TTL 死信队列)
*
*/
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 下面是设置限流机制
// 告诉他只只要我没有确认 你就不要给我 分配这个任务
/**
* 这里实际上是用在了消费端限流 这样的话
* 第一个参数 表示的是限制 本身消息的大小 如果是 0那么表示的是 不受限制
* 第二个参数
* 表示的意思是 一次性最多 接受的 消息的条目数 比如设为 2 意思这个消费者一次性只取 2 条数据
* 告诉消息队列不要一次性给消费者推送大于N个消息，要推送的前提是，前N个被推送的消息已经手动被确认，已经完成消费之后才能推送下一N个消息
* 第三个参数 表示的是 是否将上面的两个设置应用应用于 整个 通道
* false 表示的是 只是当前这个消费者自己适用
* true 表示将上面的设置应用于整个通道的消费者
*/
// channel.basicQos(1); // 一次性只是接受一个消息 没有确定之前 这里是不会给分配任务的
// channel.basicQos(0, 2, false);
// 申明一个消费者
DefaultConsumer consumer = new DefaultConsumer(channel) {
/**
*
* @param consumerTag ： 表示的是 消费者 唯一的标记
* @param envelope：信封 : 消息的一个封装
* @param properties：前面发送消息的时候 写入的键值对
* @param body ：获取到的消息的内容
* @throws IOException
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("第二个队列接受到的消息是" + new String(body));
/**
* 第一个参数是当前这条消息的标记
* 第二个参数是是否自动应答
* 在这里进行手动确认
*/
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
// 将消费者和队列进行绑定
/**
* 第一个参数：队列的名字
* 第二个参数：是否自动应答
* 第三个参数是 消费者
*/
channel.basicConsume(QUEUE_NAME, false, consumer);
}
}
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
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
最近更新: 2025/07/30, 15:37:56
