版权声明：LeifChen原创，转载请注明出处，谢谢~ https://blog.csdn.net/leifchen90/article/details/84349371

文章目录

• SpringBoot 集成 AMQP 连接 RabbitMQ
• 集成 SpringBoot
• 1 依赖项
• 2 配置文件
• 3 基本模型
• 4 生产端
• 5 消费端

参考

SpringBoot 集成 AMQP 连接 RabbitMQ

入门 RabbitMQ 消息中间件并集成 SpringBoot 2.x

集成 SpringBoot

1 依赖项

buildscript {
ext {
springBootVersion = '2.1.0.RELEASE'
}
repositories {
mavenCentral()
}
dependencies {
classpath("org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}")
}
}

apply plugin: 'org.springframework.boot'
apply plugin: 'io.spring.dependency-management'

dependencies {
// SpringBoot
implementation('org.springframework.boot:spring-boot-starter-web')

// RabbitMQ
implementation('org.springframework.boot:spring-boot-starter-amqp')

// SpringBoot Test
testImplementation('org.springframework.boot:spring-boot-starter-test')
}

2 配置文件

spring:
rabbitmq:
addresses: localhost:5672
username: guest
password: guest
virtual-host: /
connection-timeout: 15s
listener:
simple:
acknowledge-mode: manual
concurrency: 5
max-concurrency: 10
prefetch: 1
publisher-confirms: true
publisher-returns: true
template:
mandatory: true

3 基本模型

import lombok.Getter;
import lombok.Setter;
import lombok.ToString;

import java.io.Serializable;

/**
* 订单
*
* @Author LeifChen
* @Date 2018-11-20
*/
@Getter
@Setter
@ToString
public class Order implements Serializable {

private static final long serialVersionUID = 1L;

private Long id;

private String name;

private String messageId;

public Order() {
}

public Order(String name, String messageId) {
this.name = name;
this.messageId = messageId;
}
}

4 生产端

import com.chen.amqp.model.Order;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

/**
* 订单：生产者
*
* @Author LeifChen
* @Date 2018-11-20
*/
@Slf4j
@Component
public class OrderProducer {

private final RabbitTemplate rabbitTemplate;

public OrderProducer(RabbitTemplate rabbitTemplate) {
this.rabbitTemplate = rabbitTemplate;
}

/**
* 发送消息
*
* @param order
*/
public void send(Order order) {
CorrelationData correlationData = new CorrelationData(order.getMessageId());
rabbitTemplate.convertAndSend("order-exchange", "order.100", order, correlationData);
}
}

5 消费端

import com.chen.amqp.model.Order;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Map;

/**
* 订单：消费者
*
* @Author LeifChen
* @Date 2018-11-20
*/
@Component
public class OrderConsumer {

@RabbitListener(bindings = @QueueBinding(
value = @Queue(value = "order-queue", durable = "true"),
exchange = @Exchange(value = "order-exchange", type = "topic"),
key = "order.#"
))
@RabbitHandler
public void onOrderMessage(@Payload Order order, @Headers Map<String, Object> headers, Channel channel) throws IOException {
System.out.println("-----收到消息-----");
System.out.println("订单Name：" + order.getName());
System.out.println("订单MessageId：" + order.getMessageId());

Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
// ACK
channel.basicAck(deliveryTag, false);
}
}

参考

1. GitHub
2. RabbitMQ消息中间件极速入门与实战

阅读更多