SpringBoot集成RabbitMQ消息队列搭建与ACK消息确认入门

2022-07-20 16:09:55

1.RabbitMQ介绍

RabbitMQ是实现AMQP（高级消息队列协议）的消息中间件的一种，最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。RabbitMQ主要是为了实现系统之间的双向解耦而实现的。当生产者大量产生数据时，消费者无法快速消费，那么需要一个中间层。保存这个数据。

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。

RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

2.AmqpTemplate,RabbitTemplate

Spring AMQP提供了一个发送和接收消息的操作模板类AmqpTemplate。 AmqpTemplate它定义包含了发送和接收消息等的一些基本的操作功能。RabbitTemplate是AmqpTemplate的一个实现。

RabbitTemplate支持消息的确认与返回，为了返回消息，RabbitTemplate 需要设置mandatory 属性为true,并且CachingConnectionFactory 的publisherReturns属性也需要设置为true。返回的消息会根据它注册的RabbitTemplate.ReturnCallback setReturnCallback 回调发送到给客户端，

一个RabbitTemplate仅能支持一个ReturnCallback 。

为了确认Confirms消息, CachingConnectionFactory 的publisherConfirms 属性也需要设置为true，确认的消息会根据它注册的RabbitTemplate.ConfirmCallback setConfirmCallback回调发送到给客户端。一个RabbitTemplate也仅能支持一个ConfirmCallback.

3.SpringBoot集成RabbitMQ

pom.xml

<?xml version="1.0" encoding="UTF-8"?>

<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>com.example</groupId>

<artifactId>demo</artifactId>

<version>0.0.1-SNAPSHOT</version>

<packaging>jar</packaging>

<name>rabbitMQ</name>

<description>Demo project for Spring Boot</description>

<parent>

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter-parent</artifactId>

<version>2.0.2.RELEASE</version>

<relativePath/> <!-- lookup parent from repository -->

</parent>

<properties>

<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>

<java.version>1.8</java.version>

</properties>

<dependencies>

<dependency>

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter-amqp</artifactId>

</dependency>

<dependency>

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter-web</artifactId>

</dependency>

<dependency>

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter-test</artifactId>

<scope>test</scope>

</dependency>

</dependencies>

<build>

<plugins>

<plugin>

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-maven-plugin</artifactId>

</plugin>

</plugins>

</build>

</project>

自动配置信息这里我开启ACK消息确认

server.port=8083

#服务器配置

spring.application.name=rabbitmq-hello-sending

#rabbitmq连接参数

spring.rabbitmq.host=localhost

spring.rabbitmq.port=5672

spring.rabbitmq.username=linpeng

spring.rabbitmq.password=123456

# 开启发送确认

spring.rabbitmq.publisher-confirms=true

# 开启发送失败退回

spring.rabbitmq.publisher-returns=true

# 开启ACK

spring.rabbitmq.listener.direct.acknowledge-mode=manual

spring.rabbitmq.listener.simple.acknowledge-mode=manual

创建消息队列队列名：hello 和 helloObj

package com.example.demo;

import com.rabbitmq.client.Channel;

import org.springframework.amqp.core.*;

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;

@Configuration

public class RabbitConfig {

@Bean

public Queue QueueA() {

return new Queue("hello");

}

@Bean

public Queue QueueB() {

return new Queue("helloObj");

}

/**

* Fanout 就是我们熟悉的广播模式或者订阅模式，给Fanout交换机发送消息，绑定了这个交换机的所有队列都收到这个消息。

* @return

*/

@Bean

FanoutExchange fanoutExchange() {

return new FanoutExchange("ABExchange");

}

@Bean

Binding bindingExchangeA(Queue QueueA, FanoutExchange fanoutExchange) {

return BindingBuilder.bind(QueueA).to(fanoutExchange);

}

@Bean

Binding bindingExchangeB(Queue QueueB, FanoutExchange fanoutExchange) {

return BindingBuilder.bind(QueueB).to(fanoutExchange);

}

}

消息发送者 Sender 使用 RabbitTemplate 不采用 AmqpTemplate

package com.example.demo;

import org.springframework.amqp.core.AmqpTemplate;

import org.springframework.amqp.core.Message;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.amqp.rabbit.support.CorrelationData;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Component;

import org.springframework.stereotype.Service;

import java.util.Date;

//RabbitTemplate.ConfirmCallback

@Service

public class HelloSender implements RabbitTemplate.ReturnCallback {

@Autowired

// private AmqpTemplate rabbitTemplate;

private RabbitTemplate rabbitTemplate;

public void send() {

String context = "你好现在是 " + new Date() +"";

System.out.println("HelloSender发送内容 : " + context);

// this.rabbitTemplate.setConfirmCallback(this);

this.rabbitTemplate.setReturnCallback(this);

this.rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {

if (!ack) {

System.out.println("HelloSender消息发送失败" + cause + correlationData.toString());

} else {

System.out.println("HelloSender 消息发送成功 ");

}

});

this.rabbitTemplate.convertAndSend("hello", context);

}

public void sendObj() {

MessageObj obj = new MessageObj();

obj.setACK(false);

obj.setId(123);

obj.setName("zhangsan");

obj.setValue("data");

System.out.println("发送 : " + obj);

this.rabbitTemplate.convertAndSend("helloObj", obj);

}

@Override

public void returnedMessage(Message message, int i, String s, String s1, String s2) {

System.out.println("sender return success" + message.toString()+"==="+i+"==="+s1+"==="+s2);

}

}

消息接受者 Receiver 注解方式接受消息

package com.example.demo;

import com.rabbitmq.client.Channel;

import org.springframework.amqp.core.Message;

import org.springframework.amqp.core.Queue;

import org.springframework.amqp.rabbit.annotation.RabbitHandler;

import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.amqp.rabbit.annotation.RabbitListenerConfigurer;

import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;

import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistrar;

import org.springframework.amqp.support.AmqpHeaders;

import org.springframework.beans.factory.annotation.Configurable;

import org.springframework.context.annotation.Bean;

import org.springframework.messaging.handler.annotation.Headers;

import org.springframework.stereotype.Component;

import java.io.IOException;

import java.util.Date;

import java.util.Map;

@Component

@RabbitListener(queues = "hello")

public class HelloReceiver {

@RabbitHandler

public void process(String hello,Channel channel, Message message) throws IOException {

System.out.println("HelloReceiver收到 : " + hello +"收到时间"+new Date());

try {

//告诉服务器收到这条消息 已经被我消费了 可以在队列删掉 这样以后就不会再发了 否则消息服务器以为这条消息没处理掉 后续还会在发

channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

System.out.println("receiver success");

} catch (IOException e) {

e.printStackTrace();

//丢弃这条消息

//channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,false);

System.out.println("receiver fail");

}

}

}

备注：我们用注解的方式来接受消息就不要用自己创建对象实现ChannelAwareMessageListener的方式来接受消息这种方式还要去全局里面配置麻烦，直接用@RabbitListener(queues = "hello")最简单

消息确认因为我在属性配置文件里面开启了ACK确认所以如果代码没有执行ACK确认你在RabbitMQ的后台会看到消息会一直留在队列里面未消费掉只要程序一启动开始接受该队列消息的时候又会收到

channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

//消息的标识，false只确认当前一个消息收到，true确认所有consumer获得的消息

channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

//ack返回false，并重新回到队列，api里面解释得很清楚

channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);

//拒绝消息

channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);

TestController测试

@Autowired

private HelloSender helloSender;

/**

* 单生产者-单个消费者

*/

@RequestMapping("/test")

public void hello() throws Exception {

helloSender.send();

}

码农公寓

相关文章