消息中间件——RabbitMQ（十一）RabbitMQ整合Spring Cloud Stream实战！_rabbitmq与springcloud stream

最后

这份《“java高分面试指南”-25分类227页1000+题50w+字解析》同样可分享给有需要的朋友，感兴趣的伙伴们可挑战一下自我，在不看答案解析的情况，测试测试自己的解题水平，这样也能达到事半功倍的效果！（好东西要大家一起看才香）

本文已被CODING开源项目：【一线大厂Java面试题解析+核心总结学习笔记+最新讲解视频+实战项目源码】收录

需要这份系统化的资料的朋友，可以点击这里获取

3、定义RabbitmqSender类

//启动这个绑定

@EnableBinding(Barista.class)

@Service //注入到spring容器

public class RabbitmqSender {

//注入Barista

@Autowired

private Barista barista;

// 发送消息

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

try{

MessageHeaders mhs = new MessageHeaders(properties);

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

boolean sendStatus = barista.logoutput().send(msg);

System.err.println(“--------------sending -------------------”);

System.out.println(“发送数据：” + message + ",sendStatus: " + sendStatus);

}catch (Exception e){

System.err.println(“-------------error-------------”);

e.printStackTrace();

throw new RuntimeException(e.getMessage());

}

return null;

}

}

4、application.properties

server.port=8001

server.servlet.context-path=/producer

spring.application.name=producer

spring.cloud.stream.bindings.output_channel.destination=exchange-3

group相当于RabbitMQ中Queue的名称

spring.cloud.stream.bindings.output_channel.group=queue-3

##以下为集群环境配置，rabbit_cluster与下面的spring.cloud.stream.binders.rabbit_cluster是对应的。

spring.cloud.stream.bindings.output_channel.binder=rabbit_cluster

spring.cloud.stream.binders.rabbit_cluster.type=rabbit

spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.addresses=localhost:5672

spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.username=user_cp

spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.password=123456

spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.virtual-host=/vhost_cp

2.1 消费端

1、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”>

4.0.0

com.cp

rabbitmq-springcloudstream-consumer

0.0.1-SNAPSHOT

jar

rabbitmq-springcloudstream-consumer

rabbitmq-spring

org.springframework.boot

spring-boot-starter-parent

1.5.8.RELEASE

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

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

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

org.springframework.boot

spring-boot-starter-web

org.springframework.boot

spring-boot-starter

org.springframework.boot

spring-boot-starter-test

test

org.springframework.cloud

spring-cloud-starter-stream-rabbit

1.3.4.RELEASE

org.springframework.boot

spring-boot-starter-actuator

org.springframework.boot

spring-boot-maven-plugin

2、定义Barista接口

/**

• 这里的Barista接口是定义来作为后面类的参数，这一接口定义来通道类型和通道名称。

• 通道名称是作为配置用，通道类型则决定了app会使用这一通道进行发送消息还是从中接收消息。

*/

public interface Barista {

String INPUT_CHANNEL = “input_channel”;

//注解@Input声明了它是一个输入类型的通道，名字是Barista.INPUT_CHANNEL，也就是position3的input_channel。这一名字与上述配置app2的配置文件中position1应该一致，表明注入了一个名字叫做input_channel的通道，它的类型是input，订阅的主题是position2处声明的mydest这个主题

@Input(Barista.INPUT_CHANNEL)

SubscribableChannel loginput();

}

3、定义RabbitmqReceiver类

//启动binding

@EnableBinding(Barista.class)

@Service

public class RabbitmqReceiver {

@StreamListener(Barista.INPUT_CHANNEL)

public void receiver(Message message) throws Exception {

//手工签收必须要有channel与deliveryTag

Channel channel = (com.rabbitmq.client.Channel) message.getHeaders().get(AmqpHeaders.CHANNEL);

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

System.out.println(“Input Stream 1 接受数据：” + message);

System.out.println(“消费完毕------------”);

//批量签收设置为false

channel.basicAck(deliveryTag, false);

}

}

4、application.properties

server.port=8002

server.context-path=/consumer

spring.application.name=consumer

spring.cloud.stream.bindings.input_channel.destination=exchange-3

spring.cloud.stream.bindings.input_channel.group=queue-3

spring.cloud.stream.bindings.input_channel.binder=rabbit_cluster

##默认监听数

spring.cloud.stream.bindings.input_channel.consumer.concurrency=1

##针对消费端channel进行设置，是否支持requeue，重回队列

spring.cloud.stream.rabbit.bindings.input_channel.consumer.requeue-rejected=false

##是否支持签收，签收模式：手工签收

spring.cloud.stream.rabbit.bindings.input_channel.consumer.acknowledge-mode=MANUAL

##服务重连

spring.cloud.stream.rabbit.bindings.input_channel.consumer.recovery-interval=3000

##是否持久化订阅

spring.cloud.stream.rabbit.bindings.input_channel.consumer.durable-subscription=true

##最大监听数

spring.cloud.stream.rabbit.bindings.input_channel.consumer.max-concurrency=5

##采用rabbitmq方式，也可以采用kafka

spring.cloud.stream.binders.rabbit_cluster.type=rabbit

spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.addresses=localhost:5672

spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.username=user_cp

spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.password=123456

spring.cloud.stream.binders.rabbit_cluster.environment.spring.rabbitmq.virtual-host=/vhost_cp

3. 运行测试

---

3.1 运行消费端

启动项目后，查看管控台是否生成了Exchange与Queue

启动项目后，SpringCloudStream生成了Exchange与Queue。

3.2 运行生产端测试代码

@RunWith(SpringRunner.class)

@SpringBootTest

public class ApplicationTests {

@Autowired

private RabbitmqSender rabbitmqSender;

@Test

public void sendMessageTest1() throws InterruptedException {

for(int i = 0; i < 1; i ++){

try {

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

properties.put(“SERIAL_NUMBER”, “12345”);

properties.put(“BANK_NUMBER”, “abc”);

properties.put(“PLAT_SEND_TIME”, DateUtils.formatDate(new Date(), “yyyy-MM-dd HH:mm:ss.SSS”));

rabbitmqSender.sendMessage(“Hello, I am amqp sender num :” + i, properties);

} catch (Exception e) {

System.out.println(“--------error-------”);

e.printStackTrace();

}

}

TimeUnit.SECONDS.sleep(2000);

}

}

总结

无论是哪家公司，都很重视高并发高可用的技术，重视基础，重视JVM。面试是一个双向选择的过程，不要抱着畏惧的心态去面试，不利于自己的发挥。同时看中的应该不止薪资，还要看你是不是真的喜欢这家公司，是不是能真的得到锻炼。其实我写了这么多，只是我自己的总结，并不一定适用于所有人，相信经过一些面试，大家都会有这些感触。

最后我整理了一些面试真题资料，技术知识点剖析教程，还有和广大同仁一起交流学习共同进步，还有一些职业经验的分享。

本文已被CODING开源项目：【一线大厂Java面试题解析+核心总结学习笔记+最新讲解视频+实战项目源码】收录

需要这份系统化的资料的朋友，可以点击这里获取

}

TimeUnit.SECONDS.sleep(2000);

}

}

总结

无论是哪家公司，都很重视高并发高可用的技术，重视基础，重视JVM。面试是一个双向选择的过程，不要抱着畏惧的心态去面试，不利于自己的发挥。同时看中的应该不止薪资，还要看你是不是真的喜欢这家公司，是不是能真的得到锻炼。其实我写了这么多，只是我自己的总结，并不一定适用于所有人，相信经过一些面试，大家都会有这些感触。

最后我整理了一些面试真题资料，技术知识点剖析教程，还有和广大同仁一起交流学习共同进步，还有一些职业经验的分享。

[外链图片转存中…(img-AeoCEeOF-1715690317554)]

本文已被CODING开源项目：【一线大厂Java面试题解析+核心总结学习笔记+最新讲解视频+实战项目源码】收录

需要这份系统化的资料的朋友，可以点击这里获取