Spring boot集成RabbitMQ

RabbitMQ简介

RabbitMQ是一个在AMQP基础上完整的,可复用的企业消息系统
MQ全称为Message Queue, 消息队列(MQ)是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息(针对应用程序的数据)来通信,而无需专用连接来链接它们。消息传递指的是程序之间通过在消息中发送数据进行通信,而不是通过直接调用彼此来通信,直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过 队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。

AMQP就是一个协议,是一个高级抽象层消息通信协议。

虽然在同步消息通讯的世界里有很多公开标准(如 COBAR的 IIOP ,或者是 SOAP 等),但是在异步消息处理中却不是这样,只有大企业有一些商业实现(如微软的 MSMQ ,IBM 的 Websphere MQ 等),因此,在 2006 年的 6 月,Cisco 、Redhat、iMatix 等联合制定了 AMQP 的公开标准。也就是说AMQP是异步通讯的一个协议。

RabbitMQ使用场景

在项目中,将一些无需即时返回且耗时的操作提取出来,进行了异步处理,而这种异步处理的方式大大的节省了服务器的请求响应时间,从而提高了系统的吞吐量。不过大多数不仅仅是无需即时返回,甚至是执行是否成功都无所谓。如果需要即时返回则可以使用Dubbo,Spring boot与Dubbo集成可以去看Spring boot 集成Dubbox

RabbitMQ依赖

RabbitMQ并不是直接一个简单的jar包(Jar包只是提供一个基本的与RabbitMQ本身通讯的一些功能),和Dubbo相同,RabbitMQ也需要其他软件来运行,以下是RabbitMQ运行所需要的软件

1、Erlang

由于RabbitMQ软件本身是基于Erlang开发的,所以想要运行RabbitMQ必须要先按照Erlang

Erlang官网

Erlang下载地址

RabbitMQ

RabbitMQ才是实现消息队列的核心
RabbitMQ官网

RabbitMQ下载

配置RabbitMQ

安装完成后,需要完成一些配置才能使用RabbitMQ,可以直接用cmd到RabbitMQ的安装目录下的sbin目录通过命令配置,也可以直接在开始菜单中直接找到RabbitMQ Command Prompt (sbin dir)运行直接到达RabbitMQ的安装目录的sbin,为了方便,我们先启用管理插件,执行命令

rabbitmq-plugins.bat enable rabbitmq_management

即可,==注意,这是在Windows下面,如果是Linux则没有bat后缀==
然后我们添加一个用户,因为在外网环境没有用户的情况下是不能连接成功的,执行添加用户命令

rabbitmqctl.bat add_user springboot password

springboot是用户名,password是密码
然后为了方便演示,我们给springboot赋予管理员权限,方便登录管理页面

rabbitmqctl.bat set_user_tags springboot administrator

给账号赋予虚拟主机权限

rabbitmqctl.bat set_permissions -p / springboot  .* .* .*

然后启动RabbitMQ服务
访问RabbitMQ管理页面http://localhost:15672即可看见登录页面,如果没有创建用户则可以用guest,guest登录,如果有创建用户则用创建的用户登录

创建Springboot项目

因为创建spring boot项目在前面的文章已经说过很多次了,所以这里就不多说了,如果不会可以去看其他关于spring boot的博客

添加RabbitMQ相关依赖

        <!-- rabbitmq -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

每错,就是点配置,不过这样可能有点不理解,我还是把全部配置贴出来吧

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>wang.raye.rabbitmq</groupId>
  <artifactId>demo1</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <packaging>jar</packaging>

  <name>demo1</name>
  <url>http://maven.apache.org</url>

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  </properties>
<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>1.4.0.RELEASE</version>
    </parent>

    <dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>3.8.1</version>
            <scope>test</scope>
        </dependency>
        <!-- Springboot -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
            
        </dependency>
        <!-- rabbitmq -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
    
    </dependencies>
    
</project>

因为没有做其他操作,所以目前项目主要是依赖2个模块,一个Sprig boot,一个RabbitMQ

添加配置类

package wang.raye.rabbitmq.demo1;


import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * rabbitmq 的配置类
 * 
 * @author Raye
 * @since 2016年10月12日10:57:44
 */
@Configuration  
public class RabbitMQConfig {
    /** 消息交换机的名字*/
    public static final String EXCHANGE = "my-mq-exchange";
    /** 队列key1*/
    public static final String ROUTINGKEY1 = "queue_one_key1";
    /** 队列key2*/
    public static final String ROUTINGKEY2 = "queue_one_key2";

    /**
     * 配置链接信息
     * @return
     */
    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory("127.0.0.1",5672);
        
        connectionFactory.setUsername("springboot");
        connectionFactory.setPassword("password");
        connectionFactory.setVirtualHost("/");
        connectionFactory.setPublisherConfirms(true); // 必须要设置
        return connectionFactory;
    }
    
    /**  
     * 配置消息交换机
     * 针对消费者配置  
        FanoutExchange: 将消息分发到所有的绑定队列,无routingkey的概念  
        HeadersExchange :通过添加属性key-value匹配  
        DirectExchange:按照routingkey分发到指定队列  
        TopicExchange:多关键字匹配  
     */  
    @Bean  
    public DirectExchange defaultExchange() {  
        return new DirectExchange(EXCHANGE, true, false);
    } 
    /**
     * 配置消息队列1
     * 针对消费者配置  
     * @return
     */
    @Bean  
    public Queue queue() {  
       return new Queue("queue_one", true); //队列持久  
  
    }
    /**
     * 将消息队列1与交换机绑定
     * 针对消费者配置  
     * @return
     */
    @Bean  
    public Binding binding() {  
        return BindingBuilder.bind(queue()).to(defaultExchange()).with(RabbitMQConfig.ROUTINGKEY1);  
    } 
    
    /**
     * 配置消息队列2
     * 针对消费者配置  
     * @return
     */
    @Bean  
    public Queue queue1() {  
       return new Queue("queue_one1", true); //队列持久  
  
    }
    /**
     * 将消息队列2与交换机绑定
     * 针对消费者配置  
     * @return
     */
    @Bean  
    public Binding binding1() {  
        return BindingBuilder.bind(queue1()).to(defaultExchange()).with(RabbitMQConfig.ROUTINGKEY2);  
    } 
    /**
     * 接受消息的监听,这个监听会接受消息队列1的消息
     * 针对消费者配置  
     * @return
     */
    @Bean  
    public SimpleMessageListenerContainer messageContainer() {  
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());  
        container.setQueues(queue());  
        container.setExposeListenerChannel(true);  
        container.setMaxConcurrentConsumers(1);  
        container.setConcurrentConsumers(1);  
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认  
        container.setMessageListener(new ChannelAwareMessageListener() {

            public void onMessage(Message message, com.rabbitmq.client.Channel channel) throws Exception {
                byte[] body = message.getBody();  
                System.out.println("收到消息 : " + new String(body));  
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); //确认消息成功消费  
                
            }  
  
        });  
        return container;  
    }  
    /**
     * 接受消息的监听,这个监听会接受消息队列1的消息
     * 针对消费者配置  
     * @return
     */
    @Bean  
    public SimpleMessageListenerContainer messageContainer2() {  
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());  
        container.setQueues(queue1());  
        container.setExposeListenerChannel(true);  
        container.setMaxConcurrentConsumers(1);  
        container.setConcurrentConsumers(1);  
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认  
        container.setMessageListener(new ChannelAwareMessageListener() {

            public void onMessage(Message message, com.rabbitmq.client.Channel channel) throws Exception {
                byte[] body = message.getBody();  
                System.out.println("queue1 收到消息 : " + new String(body));  
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); //确认消息成功消费  
                
            }  
  
        });  
        return container;  
    }  
}

注意,为了更好的展示如何配置,我配置了2个消息队列,而本类除了链接配置哪里,其他都是针对消息消费者的,当然不管消息消费者和消息生产者都需要配置链接信息,而为了方便,所以本项目的消息消费者和生产者都在本项目,一般实际项目中不会在同一项目,由于注释很详细,我就不多说了

发送消息

为了方便发送消息,所以我直接写了一个Controller,通过访问接口的形式来调用发送消息的方法,话不多说,上代码

package wang.raye.rabbitmq.demo1;

import java.util.UUID;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * 测试RabbitMQ发送消息的Controller
 * @author Raye
 *
 */
@RestController
public class SendController implements RabbitTemplate.ConfirmCallback{
    private RabbitTemplate rabbitTemplate;
    /**
     * 配置发送消息的rabbitTemplate,因为是构造方法,所以不用注解Spring也会自动注入(应该是新版本的特性)
     * @param rabbitTemplate
     */
    public SendController(RabbitTemplate rabbitTemplate){
        this.rabbitTemplate = rabbitTemplate;
        //设置消费回调
        this.rabbitTemplate.setConfirmCallback(this);
    }
    /**
     * 向消息队列1中发送消息
     * @param msg
     * @return
     */
    @RequestMapping("send1")
    public String send1(String msg){
        String uuid = UUID.randomUUID().toString();
        CorrelationData correlationId = new CorrelationData(uuid);
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE, RabbitMQConfig.ROUTINGKEY1, msg,
                correlationId);
        return null;
    }
    /**
     * 向消息队列2中发送消息
     * @param msg
     * @return
     */
    @RequestMapping("send2")
    public String send2(String msg){
        String uuid = UUID.randomUUID().toString();
        CorrelationData correlationId = new CorrelationData(uuid);
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE, RabbitMQConfig.ROUTINGKEY2, msg,
                correlationId);
        return null;
    }
    /**
     * 消息的回调,主要是实现RabbitTemplate.ConfirmCallback接口
     * 注意,消息回调只能代表成功消息发送到RabbitMQ服务器,不能代表消息被成功处理和接受
     */
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        System.out.println(" 回调id:" + correlationData);
        if (ack) {
            System.out.println("消息成功消费");
        } else {
            System.out.println("消息消费失败:" + cause+"\n重新发送");
            
        }
    }
}

==需要注意的是消息回调只能代表消息成功发送到RabbitMQ服务器==

然后我们启动项目,访问http://localhost:8082/send1?msg=aaaa
会发现控制台输出了

收到消息 : aaaa
 回调id:CorrelationData [id=37e6e913-835a-4eca-98d1-807325c5900f]
消息成功消费

当然回调id可能不同,如果我们访问http://localhost:8082/send2?msg=bbbb 则输出

queue1 收到消息 : bbbb
 回调id:CorrelationData [id=0cec7500-3117-4aa2-9ea5-4790879812d4]
消息成功消费

最后说两句

因为本文主要是说明如何从零到springboot集成RabbitMQ,所以对于RabbitMQ的很多信息和用法没有说明,如果对RabbitMQ本身不太熟悉的可以去看看其他关于RabbitMQ的文章,附上本文demo

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 203,362评论 5 477
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 85,330评论 2 381
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 150,247评论 0 337
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,560评论 1 273
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,580评论 5 365
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,569评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 37,929评论 3 395
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,587评论 0 258
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,840评论 1 297
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,596评论 2 321
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,678评论 1 329
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,366评论 4 318
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 38,945评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,929评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,165评论 1 259
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 43,271评论 2 349
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,403评论 2 342

推荐阅读更多精彩内容

  • Spring Cloud为开发人员提供了快速构建分布式系统中一些常见模式的工具(例如配置管理,服务发现,断路器,智...
    卡卡罗2017阅读 134,585评论 18 139
  • RabbitMQ作为AMQP的代表性产品,在项目中大量使用。结合现在主流的spring boot,极大简化了开发过...
    SamHxm阅读 26,177评论 6 19
  • 安装下面是有Windows和Linux不同系统下的详细安装步骤,这了不再做介绍安装步骤这里需要注意RabbitMQ...
    RalapHao阅读 1,407评论 0 3
  • Spring Boot 参考指南 介绍 转载自:https://www.gitbook.com/book/qbgb...
    毛宇鹏阅读 46,724评论 6 342
  • RabbitMQ 即一个消息队列,主要是用来实现应用程序的异步和解耦,同时也能起到消息缓冲,消息分发的作用。 消息...
    彩虹之梦阅读 1,082评论 2 1