安裝rabbitmq
簡介: rabbitmq即一個消息隊列,主要用來實現應用程序的異步和解耦,消息緩沖,消息分發的作用.
由于rabbitmq依賴于erlang語言,所以先安裝erlang:
添加erlang solutions源
|
1
2
3
|
$ wget https://packages.erlang-solutions.com/erlang-solutions-1.0-1.noarch.rpm$ sudo rpm -uvh erlang-solutions-1.0-1.noarch.rpm$ sudo yum install erlang |
erlang完成后安裝rabbitmq:
先下載rpm：
|
1
|
wget http://www.rabbitmq.com/releases/rabbitmq-server/v3.6.6/rabbitmq-server-3.6.6-1.el7.noarch.rpm |
下載完成后安裝：
|
1
|
yum install rabbitmq-server-3.6.6-1.el7.noarch.rpm |
安裝時如果遇到下面的依賴錯誤
error: package: socat-1.7.2.3-1.el6.x86_64 (epel)
requires: libreadline.so.5()(64bit)
可以嘗試先執行
|
1
|
$ sudo yum install socat |
rabbitmq的基本操作:
|
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
|
$ sudo chkconfig rabbitmq-server on # 添加開機啟動rabbitmq服務 $ sudo /sbin/service rabbitmq-server start # 啟動服務 $ sudo /sbin/service rabbitmq-server status # 查看服務狀態 $ sudo /sbin/service rabbitmq-server stop # 停止服務 #查看當前所有用戶 $ sudo rabbitmqctl list_users #查看默認guest用戶的權限 $ sudo rabbitmqctl list_user_permissions guest #由于rabbitmq默認的賬號用戶名和密碼都是guest。為了安全起見, 先刪掉默認用戶 $ sudo rabbitmqctl delete_user guest #添加新用戶 $ sudo rabbitmqctl add_user username password #設置用戶tag $ sudo rabbitmqctl set_user_tags username administrator #賦予用戶默認vhost的全部操作權限 $ sudo rabbitmqctl set_permissions -p / username ".*" ".*" ".*"#查看用戶的權限 $ sudo rabbitmqctl list_user_permissions username |
開啟遠程訪問
默認情況下，rabbitmq的默認的guest用戶只允許本機訪問， 如果想讓guest用戶能夠遠程訪問的話，只需要將配置文件中的loopback_users列表置為空即可，如下：
|
1
|
{loopback_users, []} |
另外關于新添加的用戶，直接就可以從遠程訪問的，如果想讓新添加的用戶只能本地訪問，可以將用戶名添加到上面的列表, 如只允許admin用戶本機訪問。
|
1
|
{loopback_users, ["admin"]} |
更新配置后，重啟服務.
###springboot2.0集成rabbitmq
pom引入start依賴:
|
1
2
3
4
|
<dependency> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-starter-amqp</artifactid></dependency> |
application.properties配置:
|
1
2
3
4
5
6
7
8
|
#rabbitmq config begin# rabbitmq服務器地址 （默認為127.0.0.1）spring.rabbitmq.host=127.0.0.1# rabbitmq服務器連接端口 （默認為5672）spring.rabbitmq.port=5672spring.rabbitmq.username=adminspring.rabbitmq.password=123456#rabbitmq config end |
rabbitmq javabean配置:
|
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
|
package com.snow.config;import org.springframework.amqp.core.queue;import org.springframework.amqp.rabbit.config.simplerabbitlistenercontainerfactory;import org.springframework.amqp.rabbit.connection.connectionfactory;import org.springframework.boot.autoconfigure.amqp.simplerabbitlistenercontainerfactoryconfigurer;import org.springframework.context.annotation.bean;import org.springframework.context.annotation.configuration; /** * @description: rabbitconfig * @author: 愛飄de小子 * @createdate: 2018/8/15 19:59 * @version: 1.0 */@configurationpublic class rabbitconfig { /** * 消費者數量，默認10 */ public static final int default_concurrent = 10; /** * 每個消費者獲取最大投遞數量 默認50 */ public static final int default_prefetch_count = 50; /** * 注入 queue * @return */ @bean public queue queue() { return new queue("hello"); } /** * 并發消費配置 * @param configurer * @param connectionfactory * @return */ @bean("pointtaskcontainerfactory") public simplerabbitlistenercontainerfactory pointtaskcontainerfactory(simplerabbitlistenercontainerfactoryconfigurer configurer, connectionfactory connectionfactory) { simplerabbitlistenercontainerfactory factory = new simplerabbitlistenercontainerfactory(); factory.setprefetchcount(default_prefetch_count); factory.setconcurrentconsumers(default_concurrent); configurer.configure(factory, connectionfactory); return factory; } } |
發送消息:
|
1
2
3
4
5
6
7
8
9
10
11
12
|
/** * 注入amqptemplate */@autowiredprivate amqptemplate rabbittemplate; public void sendmessage(){ //發送消息 this.rabbittemplate.convertandsend("hello","你好,rabbitmq");} |
convertandsend方法的第一個參數為queuename,第二個參數為消息的內容
接收消息:
|
1
2
3
4
5
6
7
8
|
/** * 消息接受 * @param message */ @rabbitlistener(queues = "hello") //監聽器監聽指定的queuename public void receive(string message) { system.out.println("接收消息:" + message); } |
并發消費:
sendmessage() 發送消息后,receive() 接受消息,此時receive()接受消息,等待處理完成后,下一個消息才能進入receive(),如果想要消息異步消費,還需配置并發消費:
|
1
2
3
4
5
6
7
8
|
/** * 消息接受 并發消費 * @param message */ @rabbitlistener(queues = "hello",containerfactory = "pointtaskcontainerfactory") public void receive(string message) { system.out.println("接收消息:" + message); } |
@rabbitlistener注解中的containerfactory 是rabbitconfig配置的pointtaskcontainerfactory,可以自定義如下參數:
|
1
2
3
4
5
6
7
8
9
|
/** * 消費者數量，默認10 */ public static final int default_concurrent = 10; /** * 每個消費者獲取最大投遞數量 默認50 */ public static final int default_prefetch_count = 50; |
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持服務器之家。
原文鏈接：https://juejin.im/post/5c0273b06fb9a049f06a0021
