springboot2.0集成rabbitmq的示例代码

springboot2.0集成rabbitmq的示例代码

安装rabbitmq

简介: rabbitmq即一个消息队列,主要用来实现应用程序的异步和解耦,消息缓冲,消息分发的作用.

由于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://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

安装时如果遇到下面的依赖错误

Error: Package: socat-1.7.2.3-1.el6.x86_64 (epel)

Requires: libreadline.so.5()(64bit)

可以尝试先执行

$ sudo yum install socat

rabbitmq的基本操作:

$ sudo chkconfig rabbitmq-server on # 添加开机启动RabbitMQ服务

$ sudo /sbin/service rabbitmq-server start # 启动服务

$ sudo /sbin/service rabbitmq-server status # 查看服务状态

$ sudo /sbin/service rabbitmq-server OKzVzfXIstop # 停止服务

#查看当前所有用户

$ 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列表置为空即可，如下：

{loopback_users, []}

另外关于新添加的用户，直接就可以从远程访问的，如果想让新添加的用户只能本地访问，可以将用户名添加到上面的列表, 如只允许admin用户本机访问。

{loopback_users, ["admin"]}

更新配置后，重启服务.

###springboot2.0集成rabbitmq

pom引入start依赖:

org.springframework.boot

spring-boot-starter-amqp

application.properties配置:

#rabbitmq config begin

# rabbitmq服务器地址 （默认为127.0.0.1）

spring.rabbitmq.host=127.0.0.1

# rabbitmq服务器连接端口 （默认为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 {

/**

* 消费者数量，默认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;

}

}

发送消息:

/**

* 注入AmqpTemplate

*/

@Autowired

private AmqpTemplate rabbitTemplate;

public void sendMessage(){

//发送消息

this.rabbitTemplate.convertAndSend("hello","你好,rabbitmq");

}

convertAndSend方法的第一个参数为QueueName,第二个参数为消息的内容

接收消息:

/**

* 消息接受

* @param message

*/

@RabbitListener(queues = "hello") //监听器监听指定的QueueName

public void receive(String message) {

System.out.println("接收消息:" + message);

}

并发消费:

sendMessage() 发送消息后,receive() 接受消息,此时receive()接受消息,等待处理完成后,下一个消息才能进入receive(),如果想要消息异步消费,还需配置并发消费:

/**

* 消息接受 并发消费

* @param message

*/

@RabbitListener(queues = "hello",containerFactory = "pointTaskContainerFactory")

public void receive(String message) {

System.out.println("接收消息:" + message);

}

@RabbitListener注解中的containerFactory 是RabbitConfig配置的pointTaskContainerFactory,可以自定义如下参数:

/**

* 消费者数量，默认10

*/

public static final int DEFAULT_CONCURRENT = 10;

/**

* 每个消费者获取最大投递数量 默认50

*/

public static final int DEFAULT_PREFETCH_COUNT = 50;

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。