安裝rabbitmq
簡介: rabbitmq即一個(gè)消息隊(duì)列,主要用來實(shí)現(xiàn)應(yīng)用程序的異步和解耦,消息緩沖,消息分發(fā)的作用.
由于rabbitmq依賴于erlang語言,所以先安裝erlang:
添加erlang solutions源
$ 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：
wget http://www.rabbitmq.com/releases/rabbitmq-server/v3.6.6/rabbitmq-server-3.6.6-1.el7.noarch.rpm
下載完成后安裝：
yum install rabbitmq-server-3.6.6-1.el7.noarch.rpm
安裝時(shí)如果遇到下面的依賴錯(cuò)誤
Error: Package: socat-1.7.2.3-1.el6.x86_64 (epel)
Requires: libreadline.so.5()(64bit)
可以嘗試先執(zhí)行
$ sudo yum install socat
rabbitmq的基本操作:
$ sudo chkconfig rabbitmq-server on # 添加開機(jī)啟動RabbitMQ服務(wù)
$ sudo /sbin/service rabbitmq-server start # 啟動服務(wù)
$ sudo /sbin/service rabbitmq-server status # 查看服務(wù)狀態(tài)
$ sudo /sbin/service rabbitmq-server stop # 停止服務(wù)
# 查看當(dāng)前所有用戶
$ sudo rabbitmqctl list_users
# 查看默認(rèn)guest用戶的權(quán)限
$ sudo rabbitmqctl list_user_permissions guest
# 由于RabbitMQ默認(rèn)的賬號用戶名和密碼都是guest搜囱。為了安全起見, 先刪掉默認(rèn)用戶
$ sudo rabbitmqctl delete_user guest
# 添加新用戶
$ sudo rabbitmqctl add_user username password
# 設(shè)置用戶tag
$ sudo rabbitmqctl set_user_tags username administrator
# 賦予用戶默認(rèn)vhost的全部操作權(quán)限
$ sudo rabbitmqctl set_permissions -p / username ".*" ".*" ".*"
# 查看用戶的權(quán)限
$ sudo rabbitmqctl list_user_permissions username
開啟遠(yuǎn)程訪問
默認(rèn)情況下忍燥，RabbitMQ的默認(rèn)的guest用戶只允許本機(jī)訪問， 如果想讓guest用戶能夠遠(yuǎn)程訪問的話跟压，只需要將配置文件中的loopback_users列表置為空即可，如下：
{loopback_users, []}
另外關(guān)于新添加的用戶，直接就可以從遠(yuǎn)程訪問的，如果想讓新添加的用戶只能本地訪問壹店，可以將用戶名添加到上面的列表, 如只允許admin用戶本機(jī)訪問。
{loopback_users, ["admin"]}
更新配置后芝加，重啟服務(wù).
springboot2.0集成rabbitmq
pom引入start依賴:
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
application.properties配置:
#rabbitmq config begin
# rabbitmq服務(wù)器地址 （默認(rèn)為127.0.0.1）
spring.rabbitmq.host=127.0.0.1
# rabbitmq服務(wù)器連接端口 （默認(rèn)為5672）
spring.rabbitmq.port=5672
spring.rabbitmq.username=admin
spring.rabbitmq.password=123456
#rabbitmq config end
rabbitmq javabean配置:
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
*/
@Configuration
public class RabbitConfig {
/**
* 消費(fèi)者數(shù)量，默認(rèn)10
*/
public static final int DEFAULT_CONCURRENT = 10;
/**
* 每個(gè)消費(fèi)者獲取最大投遞數(shù)量 默認(rèn)50
*/
public static final int DEFAULT_PREFETCH_COUNT = 50;
/**
* 注入 Queue
* @return
*/
@Bean
public Queue Queue() {
return new Queue("hello");
}
/**
* 并發(fā)消費(fèi)配置
* @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;
}
}
發(fā)送消息:
/**
* 注入AmqpTemplate
*/
@Autowired
private AmqpTemplate rabbitTemplate;
public void sendMessage(){
//發(fā)送消息
this.rabbitTemplate.convertAndSend("hello","你好,rabbitmq");
}
convertAndSend方法的第一個(gè)參數(shù)為QueueName,第二個(gè)參數(shù)為消息的內(nèi)容
接收消息:
/**
* 消息接受
* @param message
*/
@RabbitListener(queues = "hello") //監(jiān)聽器監(jiān)聽指定的QueueName
public void receive(String message) {
System.out.println("接收消息:" + message);
}
并發(fā)消費(fèi):
sendMessage() 發(fā)送消息后,receive() 接受消息,此時(shí)receive()接受消息,等待處理完成后,下一個(gè)消息才能進(jìn)入receive(),如果想要消息異步消費(fèi),還需配置并發(fā)消費(fèi):
/**
* 消息接受 并發(fā)消費(fèi)
* @param message
*/
@RabbitListener(queues = "hello",containerFactory = "pointTaskContainerFactory")
public void receive(String message) {
System.out.println("接收消息:" + message);
}
@RabbitListener注解中的containerFactory 是RabbitConfig配置的pointTaskContainerFactory,可以自定義如下參數(shù):
/**
* 消費(fèi)者數(shù)量射窒，默認(rèn)10
*/
public static final int DEFAULT_CONCURRENT = 10;
/**
* 每個(gè)消費(fèi)者獲取最大投遞數(shù)量 默認(rèn)50
*/
public static final int DEFAULT_PREFETCH_COUNT = 50;
