Spring Aws动态信息未按顺序使用

Question

我正在用1个shrad将100条消息推送到流媒体。

spring:
  cloud:
    stream:
      bindings:
        myOutBound:
          destination: my-stream
          contentType: application/json

我正在循环推送消息以进行测试

@EnableBinding(MyBinder.class)
public class MyProcessor {

  @Autowired
  private MyBinder myBinder;

  public void processRollup() {
    List<MyObject> myObjects =  IntStream.range(1, 100)
        .mapToObj(Integer::valueOf)
        .map(s-> new MyObject(s))
        .collect(toList());
    myObjects.forEach(messagePayload ->{
      System.out.println(messagePayload.getId());
      myBinder.myOutBound()
          .send(MessageBuilder.withPayload(messagePayload)
              .build());
        }
    );
  }

}

我正在消费如下消息

spring:
  cloud:
    stream:
      bindings:
        RollUpInboundStream:
          group: my-consumer-group
          destination: my-stream
          content-type: application/json

消息使用未排序。

我错过了什么吗？

Answer 1

有几件事需要考虑。首先，Binder中的生产者默认基于异步模式的Kinesis MessageHandler：

messageHandler.setSync(producerProperties.getExtension().isSync());

因此，即使它寻找您以正确的顺序一个接一个地发送这些消息，这并不意味着它们以相同的顺序到达AWS上的流。

此外，即使您以同步模式发送它们，也不能保证它们以相同的顺序在AWS上结算。

请参见此处：亚马逊运动和保证订购

此外，您还可以通过显式的sequenceNumber在同一切分中实现订单保证：

要保证严格增加排序，请串行写入分片并使用SequenceNumberForOrding参数。

https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html

不幸的是，Kinesis活页夹目前不支持该选项，但我们可以通过显式AwsHeaders来克服它。在将消息发送到活页夹的输出目的地之前，在消息中设置序列号：

String sequenceNumber = messageHeaders.get(AwsHeaders.SEQUENCE_NUMBER, String.class);
    if (!StringUtils.hasText(sequenceNumber) && this.sequenceNumberExpression != null) {
        sequenceNumber = this.sequenceNumberExpression.getValue(getEvaluationContext(), message, String.class);
    }