二丶SpringBoot整合RabbitMQ

编程

什么是RabbitMQ?

rabbitMQ是一个开源的消息代理和队列的服务器,用来通过普通的协议在完全不同的应用之前共享数据,rabbitmq是Erlang语言编写的,并且rabbitMQ是基于AMQP协议的

 

1.创建product

 

<parent>

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

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

<version>2.1.5.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-web</artifactId>

</dependency>

<dependency>

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

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

<scope>test</scope>

</dependency>

<!-- springboot rabbitmq(amqp) -->

<dependency>

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

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

</dependency>

</dependencies>

 

server.servlet.context-path=/

server.port=8001

spring.rabbitmq.addresses=192.168.12.128:5672

spring.rabbitmq.username=guest

spring.rabbitmq.password=guest

spring.rabbitmq.virtual-host=/

spring.rabbitmq.connection-timeout=15000

## 使用启用消息确认模式

spring.rabbitmq.publisher-confirms=true

## 设置return消息模式,注意要和mandatory一起去配合使用

##spring.rabbitmq.publisher-returns=true

##spring.rabbitmq.template.mandatory=true

spring.application.name=rabbit-producer

spring.http.encoding.charset=UTF-8

spring.jackson.date-format=yyyy-MM-dd HH:mm:ss

spring.jackson.time-zone=GMT+8

spring.jackson.default-property-inclusion=NON_NULL

 

@SpringBootApplication

public class Application {

public static void main(String[] args) {

SpringApplication.run(Application.class, args);

}

}

package com.bfxy.rabbit.producer.component;

import org.springframework.amqp.AmqpException;

import org.springframework.amqp.core.MessagePostProcessor;

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

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

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

import org.springframework.messaging.Message;

import org.springframework.messaging.MessageHeaders;

import org.springframework.messaging.support.MessageBuilder;

import org.springframework.stereotype.Component;

import java.util.Map;

import java.util.UUID;

/**

* @auther daiison

* @create 2020-04-2020/4/26-10:30 上午

**/

@Component

public class RabbitSender {

@Autowired

private RabbitTemplate rabbitTemplate;

/**

* 这里是确认消息的回调监听接口

* 用于确认消息是否被broker所收到

*/

final RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {

/**

*

* @param correlationData 作为唯一的标识

* @param b

* @param s

* ack brober 是否罗盘成功

* cause 失败一些异常信息

*/

@Override

public void confirm(CorrelationData correlationData, boolean b, String s) {

System.err.println("ack结果为:" + b+","+correlationData);

}

};

/**

* 对外发送消息的方法

*

* @param message 具体的消息

* @param properties 附加额外的消息

* @throws Exception

*/

public void send(Object message, Map<String, Object> properties) throws Exception {

MessageHeaders mhs = new MessageHeaders(properties);

Message<?> msg = MessageBuilder.createMessage(message, mhs);

//消息确认模式

rabbitTemplate.setConfirmCallback(confirmCallback);

// 唯一的id

CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString().trim());

//发送消息

rabbitTemplate.convertAndSend("exchange-1",

"springboot.rabbit",

msg,

new MessagePostProcessor() {

@Override

public org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message) throws AmqpException {

System.err.println("--->post to do:" + message);

return message;

}

},

correlationData);

}

}

 

test测试

 

package com.bfxy.rabbit.producer;

import com.bfxy.rabbit.producer.component.RabbitSender;

import org.junit.Test;

import org.junit.runner.RunWith;

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

import org.springframework.boot.test.context.SpringBootTest;

import org.springframework.test.context.junit4.SpringRunner;

import java.awt.*;

import java.util.HashMap;

import java.util.Map;

/**

* @auther daiison

* @create 2020-04-2020/4/26-11:48 上午

**/

@RunWith(SpringRunner.class)

@SpringBootTest

public class ApplicationTest {

@Autowired

private RabbitSender rabbitSender;

@Test

public void testSender() throws Exception {

Map<String, Object> properties = new HashMap<String, Object>();

properties.put("attr1", "12345");

properties.put("attr2", "abcde");

rabbitSender.send("hello rabbitmq", properties);

Thread.sleep(10000);

}

}

 

 

 

2. 创建consumer

 

 

<parent>

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

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

<version>2.1.5.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-web</artifactId>

</dependency>

<dependency>

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

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

<scope>test</scope>

</dependency>

<!-- springboot rabbitmq(amqp) -->

<dependency>

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

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

</dependency>

</dependencies>

</project>

 

 

server.servlet.context-path=/

server.port=8002

spring.rabbitmq.addresses=192.168.12.128:5672

spring.rabbitmq.username=guest

spring.rabbitmq.password=guest

spring.rabbitmq.virtual-host=/

spring.rabbitmq.connection-timeout=15000

## 使用启用消息确认模式

## 表示消费者消费成功以后需要手工进行签收(ack) 默认为auto

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

spring.rabbitmq.listener.simple.concurrency=5

spring.rabbitmq.listener.simple.max-concurrency=10

spring.rabbitmq.listener.simple.prefetch=1

## 设置return消息模式,注意要和mandatory一起去配合使用

##spring.rabbitmq.publisher-returns=true

##spring.rabbitmq.template.mandatory=true

spring.application.name=rabbit-producer

spring.http.encoding.charset=UTF-8

spring.jackson.date-format=yyyy-MM-dd HH:mm:ss

spring.jackson.time-zone=GMT+8

spring.jackson.default-property-inclusion=NON_NULL

 

 

@SpringBootApplication

public class Application {

public static void main(String[] args) {

SpringApplication.run(Application.class, args);

}

}

 

package com.bfxy.rabbit.consumer.component;

import org.springframework.messaging.Message;

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

import org.springframework.amqp.support.AmqpHeaders;

import org.springframework.stereotype.Component;

import com.rabbitmq.client.Channel;

/**

* @auther daiison

* @create 2020-04-2020/4/26-11:08 上午

**/

@Component

public class RabbitReceive {

/**

* 组合使用,消息监听

*

* @param message

* @param channel

* @throws Exception

*/

@RabbitListener(bindings = @QueueBinding(

value = @Queue(value = "queue-1", durable = "true"),

exchange = @Exchange(name = "exchange-1", durable = "true", type = "topic", ignoreDeclarationExceptions = "true"),

key = "springboot.*"

)

)

@RabbitHandler

public void onMessage(Message message, Channel channel) throws Exception {

//1.收到消息之后进行业务端消费处理

System.err.println("----------------");

System.err.println("----- 消费消息------:" + message.getPayload());

//2.处理成功之后,获取deliveryTag 进行手动的ack 签收处理

// application 配置的手动签收模式

Long deliveryTag = (Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);

//手动签收 ack

channel.basicAck(deliveryTag, false);

}

}

 

以上是 二丶SpringBoot整合RabbitMQ 的全部内容, 来源链接: utcz.com/z/515891.html

回到顶部