11---
22layout : post
3- title : RabbitMQ_learn
3+ title : RabbitMQ的一些业务使用总结
44categories : [blog]
55tags : [RabbitMQ]
66description :
77---
88
9- RabbitMQ_learn
9+ # 一些业务使用RabbitMQ的总结
1010
11- 1 . R.Connection create TCP connection, channels shares a connection. TCP connection is costly.
11+ [ TOC ]
1212
13- 2 . flow
13+ ## Core理解
1414
15- ``` flow
16- st=>start: Producer
17- e=>end: Consumer
18- opExchange=>operation: Exchange
19- opBlinding=>operation: Blinding
20- opQueue=>operation: Queue
21- st->opExchange->opBlinding->opQueue->e
22- ```
15+ 理解几个概念,Queue是根据Bingdings规则绑定到Exchange的,消息是发布到Exchange上的,服务器决定消息投递到哪个队列。
2316
17+ ### flow
18+
19+ ``` flow
20+ st=>start: Producer
21+ e=>end: Consumer
22+ opExchange=>operation: Exchange
23+ opBlinding=>operation: Blinding
24+ opQueue=>operation: Queue
25+ st->opExchange->opBlinding->opQueue->e
26+ ```
2427
2528basic.consumer: 订阅,消费后最近的消息后** 自动接收** 下一条消息。
2629
@@ -30,3 +33,90 @@ Consumer获得消息需要返回basic.ack,一般采用auto_ack。
3033
3134basic.reject允许Consumer拒绝消息。basic.reject=true?重发:删除
3235
36+ ### connection&channel
37+
38+ 许多channels共享一个TCP connection。单个channel的生产消费能力就十分强了,而且channel的创建代价不高,无需维护连接池,最多只需要维护connection的连接池即可。
39+
40+ 关于connection&channel的双层连接池的设计比较没用,并且因为channel的自动重连机制而难以维护,我实现过,压测结果发现没啥卵用问题不少。
41+
42+ ### Bindings
43+
44+ ```
45+ channel.queue_bind(exchange=exchange_name,
46+ queue=queue_name,
47+ routing_key='city.boardcast')
48+ ```
49+
50+
51+
52+ ### Exchange
53+
54+ #### direct exchange
55+
56+ 可以通过routingKey匹配投递到对应队列。
57+
58+ 如果相同名称的routingKey绑定了多个队列的话就相当于fanout了。
59+
60+ #### fanout exchange
61+
62+ 只能广播用
63+
64+ #### topic exchange
65+
66+ 应用最为广泛的。该exchange与direct exchange基本相同,区别是可以匹配规则。
67+
68+ 譬如说,CityProducer生产了` routing_key='city.boardcast' ` ,StateProducer生产了` routing_key='state.boardcast' ` ,如果我现在有一个queue=boardcast_queue需要接受city和state所有的消息,只需要` routing_key='*.boardcast' `
69+
70+ ```
71+ channel.queue_bind(exchange=exchange_name,
72+ queue=boardcast_queue,
73+ routing_key='*.boardcast')
74+ ```
75+
76+ routing_key按照` . ` 来分割,匹配任意字符` * ` ,匹配所有routing_key` # `
77+
78+ ### VHOST
79+
80+ 顾名思义——虚拟服务器,与RMQ服务器关系类似于OS和虚拟机。
81+
82+ vhost之间绝对隔离,不同的vhost里的exchange、queue不能通行。所以对于一个业务来说,建立一个vhost保证了安全性。
83+
84+ 默认vhost是` / ` 。
85+
86+ ## Consumer
87+
88+ 消费就两种,一种推一种拉,推就是走订阅模式,mq通过长连接推到客户端;拉就是客户端不对的rpc到server去polling。
89+
90+ ### 推示例
91+
92+ ``` java
93+ Connection connection = factory. newConnection();
94+ Channel channel = connection. createChannel();
95+ channel. basicQos(1 );
96+ channel. exchangeDeclare(getExchange(), " topic" , true );
97+ channel. queueBind(getQueueName(), getExchange(), getRoutingKey());
98+ QueueingConsumer consumer = new QueueingConsumer (channel);
99+ channel. basicConsume(getQueueName(), true , consumer);
100+ QueueingConsumer . Delivery delivery = null ;
101+ try {
102+ delivery = consumer. nextDelivery();
103+ } catch (Exception ex) {
104+ logger. error(" next delivery failed: " , ex);
105+ if (ex instanceof InterruptedException ||
106+ ex instanceof ShutdownSignalException ||
107+ ex instanceof ConsumerCancelledException ) {
108+ consumer = null ;
109+ }
110+ continue ;
111+ }
112+ // do with delivery.getBody()
113+ ```
114+
115+ ### 拉
116+
117+ ```
118+ channel.basicGet("queue_name", autoAck=true)
119+ ```
120+
121+ 退的好处是及时消费,拉的好处是可以在客户端控制速率,避免尖峰把服务打挂。
122+
0 commit comments