通过spring cloud stream活页夹kafka或spring kafka，可以只交付一次，使用哪一种

Question

我试图在Spring Boot应用程序中使用spring-cloud d-stream-binder-kafka实现一次交付。我使用的版本是：

• spring-cloud-stream-binder-kafka-core-1.2.1。发布
• spring-cloud-stream-binder-kafka-1.2.1。发布
• spring-cloud-stream-codec-1.2.2。释放spring-kafka-1.1.6。释放
• Spring-集成-kafka-2.1.0。发布
• sping-集成-core-4.3.10。发布
• 动物园管理员-3.4.8
• Kafka版本：0.10.1.1

这是我的配置（云配置）：

    spring:
      autoconfigure:
        exclude: org.springframework.cloud.netflix.metrics.servo.ServoMetricsAutoConfiguration
      kafka:
        consumer:
          enable-auto-commit: false
      cloud:
        stream:
          kafka:
            binder:
              brokers: "${BROKER_HOST:xyz-aws.local:9092}"
              headers:
                - X-B3-TraceId
                - X-B3-SpanId
                - X-B3-Sampled
                - X-B3-ParentSpanId
                - X-Span-Name
                - X-Process-Id
              zkNodes: "${ZOOKEEPER_HOST:120.211.316.261:2181,120.211.317.252:2181}"
            bindings:
              feed_platform_events_input:
                consumer:
                  autoCommitOffset: false
          binders:
            xyzkafka:
              type: kafka
          bindings:
            feed_platform_events_input:
              binder: xyzkafka
              destination: platform-events
              group: br-platform-events

我有两个主要类：FeedSink接口：

package au.com.xyz.proxy.interfaces;
import org.springframework.cloud.stream.annotation.Input; 
import org.springframework.messaging.MessageChannel;

public interface FeedSink {

String FEED_PLATFORM_EVENTS_INPUT = "feed_platform_events_input";

@Input(FeedSink.FEED_PLATFORM_EVENTS_INPUT)
MessageChannel feedlatformEventsInput();
} 

事件消费者

package au.com.xyz.proxy.consumer;

@Slf4j
@EnableBinding(FeedSink.class)
public class EventConsumer {

    public static final String SUCCESS_MESSAGE =
            "SEND-SUCCESS : Successfully sent message to platform.";
    public static final String FAULT_MESSAGE = "SOAP-FAULT Code: {}, Description: {}";
    public static final String CONNECT_ERROR_MESSAGE = "CONNECT-ERROR Error Details: {}";
    public static final String EMPTY_NOTIFICATION_ERROR_MESSAGE =
            "EMPTY-NOTIFICATION-ERROR Empty Event Received from platform";

    @Autowired
    private CapPointService service;

    @StreamListener(FeedSink.FEED_PLATFORM_EVENTS_INPUT)
    /**
     * method associated with stream to process message.
     */
    public void message(final @Payload EventNotification eventNotification,
                        final @Header(KafkaHeaders.ACKNOWLEDGMENT) Acknowledgment acknowledgment) {

        String caseMilestone = "UNKNOWN";
        if (!ObjectUtils.isEmpty(eventNotification)) {
            SysMessage sysMessage = processPayload(eventNotification);
            caseMilestone = sysMessage.getCaseMilestone();
            try {
                ClientResponse response = service.sendPayload(sysMessage);
                if (response.hasFault()) {
                    Fault faultDetails = response.getFaultDetails();
                    log.error(FAULT_MESSAGE, faultDetails.getCode(), faultDetails.getDescription());
                } else {
                    log.info(SUCCESS_MESSAGE);
                }
                acknowledgment.acknowledge();
            } catch (Exception e) {
                log.error(CONNECT_ERROR_MESSAGE, e.getMessage());
            }
        } else {
            log.error(EMPTY_NOTIFICATION_ERROR_MESSAGE);
            acknowledgment.acknowledge();
        }
    }



    private SysMessage processPayload(final EventNotification eventNotification) {
        Gson gson = new Gson();
        String jsonString =  gson.toJson(eventNotification.getData());
        log.info("Consumed message for platform events with payload : {} ", jsonString);
        SysMessage sysMessage = gson.fromJson(jsonString, SysMessage.class);
        return sysMessage;
    }
    }

我已将Kafka和spring容器的autocommit属性设置为false。如果您看到在EventConsumer类中，我在我提供服务的情况下使用了Acknowledge。sendPayload成功，没有例外。我希望容器移动偏移量并轮询下一个记录。我观察到的是：

>

场景2——如果发生异常，然后将后续消息推送到Kafka。我看到新消息被处理，偏移量被移动。这意味着我丢失了未被确认的信息。所以问题是我是否处理过这个问题。我如何控制从上次提交中读取信息，而不仅仅是最新的信息并对其进行处理。我假设在内部进行了一次民意调查，但没有考虑到或不知道最后一条信息没有得到确认。我不认为有多条线索在读Kafka。我不知道@Input和@StreamListener注释是如何控制的。我假设线程由属性使用者控制。控制线程的并发性，默认设置为1。

所以我做了研究，发现了很多链接，但不幸的是，它们都没有回答我的具体问题。我看了（https://github.com/spring-cloud/spring-cloud-stream/issues/575），里面有马吕斯的评论（https://stackoverflow.com/users/809122/marius-bogoevici）：

请注意，Kafka不提供单独的消息确认，这意味着确认转换为将最新消耗的偏移量更新为已确认消息的偏移量（每个主题/分区）。这意味着，如果您无序地确认来自同一主题分区的消息，那么一条消息可以“确认”它之前的所有消息。

当有一个线程时，不确定是否是订单问题。

很抱歉发了这么长的帖子，但我想提供足够的信息。最重要的是，我正在努力避免在使用Kafka时丢失消息，我正在尝试看看spring cloud stream binder kafka是否可以完成这项工作，或者我必须寻找替代方案。

2018年7月6日更新

我看到了这篇帖子https://github.com/spring-projects/spring-kafka/issues/431这是解决我问题的更好方法吗？我可以试试最新版本的《SpringKafka》

@KafkaListener(id = "qux", topics = "annotated4", containerFactory = "kafkaManualAckListenerContainerFactory",
                containerGroup = "quxGroup")
public void listen4(@Payload String foo, Acknowledgment ack, Consumer<?, ?> consumer) {

• 这是否有助于控制将偏移量设置为上次成功处理记录的位置？我怎样才能从listen方法中做到这一点。消费者seekToEnd（）；然后如何重置监听方法以获取该记录
• 将消费者放在签名中是否能为与消费者打交道提供支持？或者我还需要做什么
• 我应该使用Acknowledge还是consumer。commitSyncy（）
• 集装箱工厂的意义是什么。我必须把它定义为豆子吗
• 我需要@EnableKafka和@Configuration才能让上述方法发挥作用吗？请记住，该应用程序是一个Spring Boot应用程序
• 通过添加Consumer to listen方法，我不需要实现ConsumerAware接口

最后但并非最不重要的一点是，如果上述方法可行，是否可以提供一些例子。

2018年7月12日更新

谢谢加里(https://stackoverflow.com/users/1240763/gary-russell)提供使用maxAttempts的提示。我用过这种方法。我能够做到一次送达，并保持信息的秩序。

我更新的云配置：

    spring:
      autoconfigure:
        exclude: org.springframework.cloud.netflix.metrics.servo.ServoMetricsAutoConfiguration
      kafka:
        consumer:
          enable-auto-commit: false
      cloud:
        stream:
          kafka:
            binder:
              brokers: "${BROKER_HOST:xyz-aws.local:9092}"
              headers:
                - X-B3-TraceId
                - X-B3-SpanId
                - X-B3-Sampled
                - X-B3-ParentSpanId
                - X-Span-Name
                - X-Process-Id
              zkNodes: "${ZOOKEEPER_HOST:120.211.316.261:2181,120.211.317.252:2181}"
            bindings:
              feed_platform_events_input:
                consumer:
                  autoCommitOffset: false
          binders:
            xyzkafka:
              type: kafka
          bindings:
            feed_platform_events_input:
              binder: xyzkafka
              destination: platform-events
              group: br-platform-events
              consumer:
                maxAttempts: 2147483647
                backOffInitialInterval: 1000
                backOffMaxInterval: 300000
                backOffMultiplier: 2.0

事件使用者与我最初的实现保持一致。除了重新引发错误以使容器知道处理已失败之外。如果您只是捕获了它，那么容器就不可能知道消息处理失败了。通过承认。确认您只是在控制偏移量提交。为了重试，必须抛出异常。不要忘记将kafka客户机autocommit属性和spring（容器级别）autocommitOffset属性设置为false。就这样。

Answer 1

正如马吕斯所解释的，Kafka在日志中只保留一个偏移量。如果处理下一条消息，并更新偏移量；失败的消息丢失。

您可以将失败的消息发送到死信主题（将enableDlq设置为true）。

Spring Kafka（2.1.x）的最新版本具有特殊的错误处理程序Containers ToppingErrorHandler，当发生异常时，它会停止容器，而seekToCurInterrorHandler会导致失败的消息重新传递。