Spring Cloud Stream 是一个构建消息驱动微服务的框架,应用程序通过input通道或者output通道来与Spring Cloud Stream中binder交互,通过配置来binding. 而Spring Cloud Stream的binder负责与中间件交互, 消息的中间件有(RabbitMQ, Kafka, ActiveMQ).
本文以RabbitMQ作为中间件作为讲解(启动rabbitMQ: rabbitmq-plugins enable rabbitmq_management)
Producer
1.通过IDEA的spring init新建SpringBoot项目
引入依赖
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
配置yml文件
server:
port: 8081
spring:
application:
name: producer
cloud:
stream:
bindings:
output1:
destination: Theme-People #可以理解为一个和订阅者的通道
group: Group-Boy
content-type: application/json
producer:
partitionKeyExpression: headers['partitionKey']
partitionCount: 2
新建一个实体类 用于发送
package com.cloud.cloud_streaming.vo;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.ToString;
@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString
public class Person {
private String id;
private String name;
private int age;
private String password;
}
新建输出通道
package com.cloud.cloud_streaming;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.stereotype.Component;
@Component
public interface OutputInterface {
String OUTPUT1 = "output1";
String OUTPUT2 = "output2";
//创建一个输出通道 output1
@Output(OUTPUT1)
MessageChannel output1();
}
新建一个service用域处理发送消息
package com.cloud.cloud_streaming;
import com.cloud.cloud_streaming.vo.Person;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class StreamService {
@Autowired
@Qualifier(OutputInterface.OUTPUT1)
MessageChannel channel1;
public void sendMessage(String channel) {
String message = "Send Message Manually,From Channel:" + channel;
log.info( "Send Message from channel:" + channel );
switch (channel) {
case OutputInterface.OUTPUT1:
channel1.send( MessageBuilder.withPayload( message ).build() );
return;
default:
log.info( "参数错误: " + channel );
return;
}
}
}
新建一个定时任务类
package com.cloud.cloud_streaming;
import com.cloud.cloud_streaming.vo.Person;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
@Slf4j
@EnableBinding(value = {OutputInterface.class})
@EnableScheduling
public class MessageSender {
@Autowired
private OutputInterface outputInterface;
/**
* 第一种方法, 没有指定output的MessageChannel, 通过OutputInterface去拿具体的Channel
* 设置partitionKey主要是为了分区用, 可以根据根据这个partitionKey来分区
采用定时任务去发送
*/
@Scheduled(initialDelay = 1000, fixedRate = 5000)
public void sendMessageMethod1() {
Message message = MessageBuilder.withPayload( "From sendMessageMethod1" )
.setHeader( "partitionKey", 1 )
.build();
outputInterface.output1().send( message );
}
/**
* 第二种方法, 直接指定output的MessageChannel
*/
// @Autowired
// @Qualifier(OutputInterface.OUTPUT2)
// MessageChannel output;
//
// @Scheduled(initialDelay = 2000, fixedRate = 4000)
// public void sendMessageMethod2() {
// Person p = new Person();
// p.setName( "Person2" );
// p.setAge( 1 );
//
// output.send( MessageBuilder.withPayload( p )
// .setHeader( "partitionKey", 2 )
// .build() );
// }
}
controller层
package com.cloud.cloud_streaming;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@Slf4j
public class StreamController {
@Autowired
StreamService streamService;
@GetMapping("/sendMessageByChannalName")
public void sendMessage(String channel) {
streamService.sendMessage( channel );
}
}
cusomer
1.引入依赖
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
<version>2.1.2.RELEASE</version>
</dependency>
2.配置yml
spring:
application:
name: consumer
cloud:
stream:
bindings:
input1:
destination: Theme-Animal #h对应的输出通道 同一个通道
group: Group-Boy
content-type: application/json
consumer:
partitioned: true #是否分区
instance-index: 0 #用那个分区来接收消息
instance-count: 2 #分区数
输入类
package com.tg.stream_cusomer;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.stereotype.Component;
@Component
public interface InputInterface {
String INPUT1 = "input1";
@Input(INPUT1)
SubscribableChannel input1();
}
同样新建一个实体类,因为采用的是发送实体
package com.tg.stream_cusomer;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.ToString;
@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString
public class Person {
private String name;
private int age;
}
接受消息的处理层
package com.tg.stream_cusomer;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
@Slf4j
@EnableBinding(value = {InputInterface.class})
public class MessageReceiver {
@StreamListener(InputInterface.INPUT1)
public void receiveMessageFromChannel1(@Payload Message <String> payload, @Header(AmqpHeaders.CONSUMER_QUEUE) String partition) {
log.info( "ReceiveMessageFrom INPUT1, message: {}, Queue:{}", payload.getPayload(), partition );
}
}
启动producer的project
再启动cusomer的project
网友评论