springboot2.0集成rabbitmq的示例代碼

更新時(shí)間：2018年12月02日 16:31:13 作者：flyingkid

這篇文章主要介紹了springboot2.0集成rabbitmq的示例代碼，小編覺(jué)得挺不錯(cuò)的，現(xiàn)在分享給大家，也給大家做個(gè)參考。一起跟隨小編過(guò)來(lái)看看吧

安裝rabbitmq

簡(jiǎn)介: rabbitmq即一個(gè)消息隊(duì)列,主要用來(lái)實(shí)現(xiàn)應(yīng)用程序的異步和解耦,消息緩沖,消息分發(fā)的作用.

由于rabbitmq依賴于erlang語(yǔ)言,所以先安裝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 # 添加開(kāi)機(jī)啟動(dòng)RabbitMQ服務(wù)
 
$ sudo /sbin/service rabbitmq-server start # 啟動(dòng)服務(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)的賬號(hào)用戶名和密碼都是guest。為了安全起見(jiàn), 先刪掉默認(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

開(kāi)啟遠(yuǎn)程訪問(wèn)

默認(rèn)情況下，RabbitMQ的默認(rèn)的guest用戶只允許本機(jī)訪問(wèn)，如果想讓guest用戶能夠遠(yuǎn)程訪問(wèn)的話，只需要將配置文件中的loopback_users列表置為空即可，如下：

{loopback_users, []}

另外關(guān)于新添加的用戶，直接就可以從遠(yuǎn)程訪問(wèn)的，如果想讓新添加的用戶只能本地訪問(wèn)，可以將用戶名添加到上面的列表, 如只允許admin用戶本機(jī)訪問(wèn)。

{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: 愛(ài)飄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)聽(tīng)器監(jiān)聽(tīng)指定的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;

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: