springCloud stream message 开发 单元测试

1.maven依赖
2.发送消息及单元测试

2.1发送消息 使用默认提供的Source类，提供一个"Output" MessageChannel
2.2测试消息发送 使用testsupport类MessageCollector拉取消息
2.3配置内容

3.消息接收和转发及单元测试

3.1 消息处理 含接收和转发 EnableBinding Processor.class
3.2测试类
3.3配置

4.消息接收处理及参数验证测试

4.1 消息接受处理
4.2测试验证接收入参
4.3配置

5.自定义通道发送接收

5.1自定义消息发送接收通道 MySource MySink
5.2配置通道
5.3消息发送使用
5.4消息接收使用

1.maven依赖

<dependencies>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-rabbit</artifactId>
</dependency>

<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <optional>true</optional>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit-test</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-test-support</artifactId>
    <scope>test</scope>
</dependency>
</dependencies>

2.1发送消息
使用默认提供的Source类，提供一个"Output" MessageChannel

/***
 * 定时任务发送消息
 * */
@EnableScheduling
@EnableBinding(Source.class)
public class UsageDetailSender {

    @Autowired
    private Source source;

    private String[] users = {"user1", "user2", "user3", "user4", "user5"};

    @Scheduled(fixedDelay = 1000)
    public void sendEvents() {
        UsageDetail usageDetail = new UsageDetail();

        usageDetail.setUserId(this.users[new Random().nextInt(5)]);
        usageDetail.setDuration(new Random().nextInt(300));
        usageDetail.setData(new Random().nextInt(700));
        this.source.output()
            .send(MessageBuilder.withPayload(usageDetail).build());
    }
}

2.2测试消息发送
使用testsupport类MessageCollector拉取消息

@RunWith(SpringRunner.class)
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
public class UsageDetailSenderApplicationTests {

    @Autowired
    private MessageCollector messageCollector;

    @Autowired
    private Source source;

    @Test
    public void contextLoads() {
    }

    @Test
    public void testUsageDetailSender() throws Exception {
        Message message = this.messageCollector.forChannel(
                this.source.output())
                .poll(1, TimeUnit.SECONDS);
        String usageDetailJSON = message.getPayload().toString();
        assertTrue(usageDetailJSON.contains("userId"));
        assertTrue(usageDetailJSON.contains("duration"));
        assertTrue(usageDetailJSON.contains("data"));
    }
}

2.3配置内容

spring.cloud.stream.bindings.output.destination=usage-detail
spring.cloud.stream.bindings.output.producer.requiredGroups=usage-cost-consumer

3.1 消息处理 EnableBinding Processor.class

@EnableBinding(Processor.class)
public class UsageCostProcessor {

    private double ratePerSecond = 0.1;

    private double ratePerMB = 0.05;

    @StreamListener(Processor.INPUT)
    @SendTo(Processor.OUTPUT)
    public UsageCostDetail processUsageCost(UsageDetail usageDetail) {
        UsageCostDetail usageCostDetail = new UsageCostDetail();
        usageCostDetail.setUserId(usageDetail.getUserId());
        usageCostDetail.setCallCost(usageDetail.getDuration() * this.ratePerSecond);
        usageCostDetail.setDataCost(usageDetail.getData() * this.ratePerMB);
        return usageCostDetail;
    }
}

3.2测试类

@RunWith(SpringRunner.class)
@SpringBootTest
public class UsageCostProcessorRabbitApplicationTests {
@Autowired
private Processor processor;
@Autowired
private MessageCollector messageCollector;
@Test
public void testUsageCostProcessor() throws Exception {
    this.processor.input().send(MessageBuilder.withPayload("{\"userId\":\"user3\",\"duration\":101,\"data\":502}").build());
    Message message = this.messageCollector.forChannel(this.processor.output()).poll(1, TimeUnit.SECONDS);
    assertTrue(message.getPayload().toString().equals("{\"userId\":\"user3\",\"callCost\":10.100000000000001,\"dataCost\":25.1}"));
}
}

3.3配置

spring.cloud.stream.bindings.input.destination=usage-detail
spring.cloud.stream.bindings.input.group=usage-cost-consumer
spring.cloud.stream.bindings.output.destination=usage-cost
spring.cloud.stream.bindings.output.producer.requiredGroups=logger

4.1 消息接受处理

@EnableBinding(Sink.class)
public class UsageCostLogger {
private static final Logger logger = LoggerFactory.getLogger(UsageCostLoggerRabbitApplication.class);
@StreamListener(Sink.INPUT)
public void process(UsageCostDetail usageCostDetail) {
    logger.info(usageCostDetail.toString());
}
}

4.2测试验证接收入参

@RunWith(SpringRunner.class)
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
public class UsageCostLoggerRabbitApplicationTests {
@Autowired
protected Sink sink;
@Autowired
protected UsageCostLogger usageCostLogger;
@Test
public void testUsageCostLogger() throws Exception {
    ArgumentCaptor<UsageCostDetail> captor = ArgumentCaptor.forClass(UsageCostDetail.class);
    this.sink.input().send(MessageBuilder.withPayload("{\"userId\":\"user3\",\"callCost\":10.100000000000001,\"dataCost\":25.1}").build());
    //捕获usageCostLogger.process的入参
    verify(this.usageCostLogger).process(captor.capture());
    //校验入参值
    assertEquals("{\"userId\":\"user3\",\"callCost\":10.100000000000001,\"dataCost\":25.1}", captor.getValue());
}
@EnableAutoConfiguration
@EnableBinding(Sink.class)
static class TestConfig {
    // Override `UsageCostLogger` bean for spying.
    @Bean
    @Primary
    public UsageCostLogger usageCostLogger() {
        return spy(new UsageCostLogger());
    }
}
}

4.3配置

spring.cloud.stream.bindings.input.destination=usage-cost
spring.cloud.stream.bindings.input.group=logger

5.1自定义消息发送接收通道

public interface MySource {

    String OC_BILL_OUTPUT = "oc-bill-output";

    /**
     * 发布对帐单通知
     * @return 消息体
     */
    @Output(OC_BILL_OUTPUT)
    MessageChannel paymentBill();
}

public interface MySink {
    String OC_BILL_INPUT = "oc-bill-input";

    /**
     * 对帐消息接收
     * @return 消息体
     */
    @Input(OC_BILL_INPUT)
    SubscribableChannel paymentBill();
}

5.2配置通道

spring.cloud.stream.bindings.oc-bill-input.destination=payment-bill
spring.cloud.stream.bindings.oc-bill-input.content-type=application/json
spring.cloud.stream.bindings.oc-bill-input.group=pay-admin

spring.cloud.stream.bindings.oc-bill-output.destination=pay-payment-bill
spring.cloud.stream.bindings.oc-bill-output.content-type=application/json
spring.cloud.stream.bindings.oc-bill-output.group=pay-admin

5.3消息发送使用

@RestController
public class MyController {

    @Autowired
    private MySource mySource;

    /**发送对帐通知*/
    @PostMapping(value = "/send")
    public ResponseResult send(@RequestBody BillVO billVO) {

        Message<BillVO> message = MessageBuilder.withPayload(billVO).build();

        mySource.paymentBill().send(message);

        return ResponseResult.ofSuccess();

    }
}

5.4消息接收使用

@StreamListener(MySink.OC_BILL_INPUT)
public void check(@Payload BillVO billVO){

    process(billVO);
}