java - 如果正在使用不同的线程,Spring Cloud Stream 可轮询消费者 dlq 和 errorChannel 将不起作用
问题描述
为了使用带有 Kafka binder 的 Spring Cloud Stream 3.1.1 管理长时间运行的任务,我们需要使用 Pollable Consumer 在单独的线程中手动管理消费,这样 Kafka 就不会触发重新平衡。为此,我们定义了一个新注解来管理 Pollable Consumer。这种方法的问题是因为工作需要在单独的线程中进行管理,任何抛出的异常都不会最终出现在errorChannel
DLQ 中。
private final ExecutorService executor = Executors.newFixedThreadPool(1);
private volatile boolean paused = false;
@Around(value = "@annotation(pollableConsumer) && args(dataCapsule,..)")
public void handleMessage(ProceedingJoinPoint joinPoint,
PollableConsumer pollableConsumer, Object dataCapsule) {
if (dataCapsule instanceof Message) {
Message<?> message = (Message<?>) dataCapsule;
AcknowledgmentCallback callback = StaticMessageHeaderAccessor
.getAcknowledgmentCallback(message);
callback.noAutoAck();
if (!paused) {
// The separate thread is not busy with a previous message, so process this message:
Runnable runnable = () -> {
try {
paused = true;
// Call method to process this Kafka message
joinPoint.proceed();
callback.acknowledge(Status.ACCEPT);
} catch (Throwable e) {
callback.acknowledge(Status.REJECT);
throw new PollableConsumerException(e);
} finally {
paused = false;
}
};
executor.submit(runnable);
} else {
// The separate thread is busy with a previous message, so re-queue this message for later:
callback.acknowledge(Status.REQUEUE);
}
}
}
我们可以创建一个不同的输出通道来在出现异常的情况下发布消息,但感觉我们正在尝试实现一些可能没有必要的东西。
更新 1
我们添加了这些 bean:
@Bean
public KafkaTemplate<String, byte[]> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}
@Bean
public ProducerFactory<String, byte[]> producerFactory() {
Map<String, Object> configProps = new HashMap<>();
configProps.put(
org.apache.kafka.clients.producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
"http://localhost:9092");
configProps.put(
org.apache.kafka.clients.producer.ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class);
configProps.put(
org.apache.kafka.clients.producer.ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
KafkaAvroSerializer.class);
return new DefaultKafkaProducerFactory<>(configProps);
}
@Bean
public KafkaAdmin admin() {
Map<String, Object> configs = new HashMap<>();
configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "http://localhost:9092");
return new KafkaAdmin(configs);
}
@Bean
public NewTopic topicErr() {
return TopicBuilder.name("ERR").partitions(1).replicas(1).build();
}
@Bean
public SeekToCurrentErrorHandler eh(KafkaOperations<String, byte[]> template) {
return new SeekToCurrentErrorHandler(new DeadLetterPublishingRecoverer(
template,
(cr, e) -> new TopicPartition("ERR", 1)),
new FixedBackOff(0L, 1L));
}
并且enable-dlq
未设置spring.cloud.stream.kafka.bindings.channel-name.consumer
但我们仍然看不到任何消息正在生成到 ERR 主题。即使是主线程抛出的任何异常。
如果enable-dlq
设置为 true,则主线程上的异常将被发布到默认的 dlq 主题中,并且正如预期的那样,子线程上的异常将被忽略。
更新 2
加里的例子似乎在一般情况下工作。尽管我们需要进行一些修改,因为我们使用了已弃用的 StreamListner 方法而不是 Functions,但仍有一些问题我们无法通过我们的案例解决。
- 主题名称似乎始终是预期的,
channel_name+.DLT
因为我们无法弄清楚如何使用不同的名称dlq
。我们为所有消费者使用一个dlq
主题,这似乎不是 Spring-kafka 默认 DLT 所期望的。 - 看来我们需要在 DLT 上至少拥有与消费者主题相同数量的分区。否则,此解决方案不起作用。虽然这对我们来说似乎不是一个实际的假设,但不确定如何管理。
- 有没有一种方法可以利用 Spring 重试,类似于 Spring Cloud Stream 在幕后所做的?或者这需要单独实施?即基于
max.attempts
DLQ部分重试工作,然后启动。 - 我可以看到,在示例中,Spring 执行器已用于通过
this.endpoint.changeState("polled", State.PAUSED)
和更新通道状态this.endpoint.changeState("polled", State.RESUMED)
。为什么我们需要与暂停、重新排队等结合使用。不这样做的副作用是什么?
解决方案
你的观察是正确的;错误处理绑定到线程。
您可以DeadLetterPublishingRecoverer
直接在代码中使用 a 来使发布 DLQ 变得更容易(而不是输出通道)。这样,您将获得带有异常信息等的增强标头。
https://docs.spring.io/spring-kafka/docs/current/reference/html/#dead-letters
编辑
这是一个例子;我正在暂停绑定以防止在“作业”运行时进行任何新的交付,而不是像您正在做的那样重新排队交付。
@SpringBootApplication
@EnableScheduling
public class So67296258Application {
public static void main(String[] args) {
SpringApplication.run(So67296258Application.class, args);
}
@Bean
TaskExecutor exec() {
return new ThreadPoolTaskExecutor();
}
@Bean
DeadLetterPublishingRecoverer recoverer(KafkaOperations<Object, Object> template) {
return new DeadLetterPublishingRecoverer(template);
}
@Bean
NewTopic topic() {
return TopicBuilder.name("polled.DLT").partitions(1).replicas(1).build();
}
@Bean
MessageSourceCustomizer<KafkaMessageSource<?, ?>> customizer() {
return (source, dest, group) -> source.setRawMessageHeader(true);
}
}
@Component
class Handler {
private static final Logger LOG = LoggerFactory.getLogger(Handler.class);
private final PollableMessageSource source;
private final TaskExecutor exec;
private final BindingsEndpoint endpoint;
private final DeadLetterPublishingRecoverer recoverer;
Handler(PollableMessageSource source, TaskExecutor exec, BindingsEndpoint endpoint,
DeadLetterPublishingRecoverer recoverer) {
this.source = source;
this.exec = exec;
this.endpoint = endpoint;
this.recoverer = recoverer;
}
@Scheduled(fixedDelay = 5_000)
public void process() {
LOG.info("Polling");
boolean polled = this.source.poll(msg -> {
LOG.info("Pausing Binding");
this.endpoint.changeState("polled", State.PAUSED);
AcknowledgmentCallback callback = StaticMessageHeaderAccessor.getAcknowledgmentCallback(msg);
callback.noAutoAck();
// LOG.info(msg.toString());
this.exec.execute(() -> {
try {
runJob(msg);
}
catch (Exception e) {
this.recoverer.accept(msg.getHeaders().get(KafkaHeaders.RAW_DATA, ConsumerRecord.class), e);
}
finally {
callback.acknowledge();
this.endpoint.changeState("polled", State.RESUMED);
LOG.info("Resumed Binding");
}
});
});
LOG.info("" + polled);
}
private void runJob(Message<?> msg) throws InterruptedException {
LOG.info("Running job");
Thread.sleep(30_000);
throw new RuntimeException("fail");
}
}
spring.cloud.stream.pollable-source=polled
spring.cloud.stream.bindings.polled-in-0.destination=polled
spring.cloud.stream.bindings.polled-in-0.group=polled
编辑2
补充问题的答案:
1、2:查看 Spring for Apache Kafka 文档:https ://docs.spring.io/spring-kafka/docs/current/reference/html/#dead-letters
DLPR 有一个备用构造函数,使您能够指定目标解析器。默认只是追加.DLT
并使用相同的分区。javadocs 指定如何指定目标分区:
/**
* Create an instance with the provided template and destination resolving function,
* that receives the failed consumer record and the exception and returns a
* {@link TopicPartition}. If the partition in the {@link TopicPartition} is less than
* 0, no partition is set when publishing to the topic.
* @param template the {@link KafkaOperations} to use for publishing.
* @param destinationResolver the resolving function.
*/
当 时null
,KafkaProducer
选择分区。
RetryTemplate
使用适当的重试和退避策略连接 a ;然后
retryTemplate.execute(context -> { ... },
context -> {...});
第二个参数是 a RecoveryCallback
,在重试用尽时调用。
- 它更有效率。使用您的解决方案,您可以在处理上一个任务时继续检索和重新排队交付。通过暂停绑定,我们告诉 kafka 在
poll()
我们恢复消费者之前不要再发送任何记录。这允许我们通过轮询消费者来保持消费者的生命,但没有检索和重置偏移量的开销。
推荐阅读
- composer-php - 无法使用 localhost 访问超薄主页
- c# - 动态信号量请求计数
- javascript - 折叠/展开插入符号图标更改所有项目,而不仅仅是一个
- php - FastCGI 模式下的 PHP 不会读取 .user.ini 指令
- php - PHP中压缩的xml格式数据
- python - django 表单应该根据字符串比较返回数据
- ansible - 包含和标签在 ansible 中不起作用
- zingchart - 为 ZingCharts 散点图实现时间范围切片器
- scala - Shapeless 什么时候需要依赖类型?
- php - 根据下拉菜单 wordpress 按日期对帖子进行排序