工程学习台

主题

主要章节

Spring AMQP 与 Kafka 集成：监听生命周期、确认、重试与事务 ​

Boot 对消息产品的"支持"只是一个 starter：启用配置、监听生命周期、确认/异常/重试/事务这些行为属于 Spring 侧集成层，与 broker 机制（回查 MQ 可靠性、Kafka 消费位点）分属两层，不能把"自动装配成功"当交付保证。本页固定组合 Spring AMQP 4.0.6（amqp-client 5.27.1）、Spring Kafka 4.0.8（kafka-clients 4.1.2）、Boot 4.0.9 / Framework 7.0.10、Liberica JDK 21.0.11，按产品分别实跑：AMQP 用 Qpid Broker-J 10.1.1 内存 broker（AMQP 0-9-1 与 rabbit 客户端兼容的替身，不是 RabbitMQ 本体），Kafka 用真实 4.1.2 KRaft 单机（Maven 构件组装，进程由实验 runner 启停）。Spring Pulsar 2.0.8（BOM 钉版）因真实 broker 未授权下载受阻登记。实验入口在 labs/broker-integrations-u48。

Spring AMQP：声明、监听生命周期与应用侧死信 ​

启用配置三件套：CachingConnectionFactory、RabbitAdmin（声明交换机/队列/绑定）、@RabbitListener + 容器工厂。实验按顺序验证了一个真实约束：监听容器自动启动会与声明发生竞态（容器在队列声明前连接会 404），所以工厂设置 autoStartup=false，声明完成后再 container.start()——这是"启用配置"阶段最容易踩的时序坑。

监听生命周期：start 后消息被消费；stop 后消息积压在队列（passive 计数 1）；start 后恢复消费。stop/start 期间连接关闭、消费者摘除，行为由容器（SimpleMessageListenerContainer）保证，broker 只是排队。

异常与死信：Qpid 不支持 rabbit 的 x-dead-letter-exchange 扩展参数（声明即 542 错误），所以实验用 Spring 侧的应用死信——defaultRequeueRejected=false + RepublishMessageRecoverer。毒消息的处理路径在固定源码里：

原码java
	public void recover(Message message, @Nullable Throwable cause) {
		MessageProperties messageProperties = message.getMessageProperties();
		Map<String, @Nullable Object> headers = messageProperties.getHeaders();
		if (cause != null) {
			String exceptionMessage = cause.getCause() != null ? cause.getCause().getMessage() : cause.getMessage();
			@Nullable String[] processed = processStackTrace(cause, exceptionMessage);
			String stackTraceAsString = processed[0];
			String truncatedExceptionMessage = processed[1];
			if (truncatedExceptionMessage != null) {
				exceptionMessage = truncatedExceptionMessage;
			}
			if (this.includeStackTrace) {
				headers.put(X_EXCEPTION_STACKTRACE, stackTraceAsString);
			}
			headers.put(X_EXCEPTION_MESSAGE, exceptionMessage);
		}
		headers.put(X_ORIGINAL_EXCHANGE, messageProperties.getReceivedExchange());
		headers.put(X_ORIGINAL_ROUTING_KEY, messageProperties.getReceivedRoutingKey());
		Map<? extends String, ?> additionalHeaders = additionalHeaders(message, cause);
		if (additionalHeaders != null) {
			headers.putAll(additionalHeaders);
		}

		if (messageProperties.getDeliveryMode() == null) {
			messageProperties.setDeliveryMode(this.deliveryMode);
		}

		String exchangeName = this.errorExchangeNameExpression.getValue(this.evaluationContext, message, String.class);
		String rk = this.errorRoutingKeyExpression.getValue(this.evaluationContext, message, String.class);
		String routingKey = rk != null ? rk : this.prefixedOriginalRoutingKey(message);
		if (null != exchangeName) {
			doSend(exchangeName, routingKey, message);
			if (this.logger.isWarnEnabled()) {
				this.logger.warn("Republishing failed message to exchange '" + exchangeName
						+ "' with routing key " + routingKey);
			}
		}
		else {
			doSend(null, routingKey, message);
			if (this.logger.isWarnEnabled()) {
				this.logger.warn("Republishing failed message to the template's default exchange with routing key "
						+ routingKey);
			}
		}
	}

recover 把失败消息（带错误头）重新发布到配置的交换机/队列；配合无状态重试拦截器时，重试耗尽后走同一出口——实验观察：poison 一次尝试即入 DLQ，retry-me 三次尝试后入 DLQ。注意语义差别：broker 侧 DLX 是投递拓扑的一部分，应用侧 republish 是一次普通的 publisher 发送——后者要考虑发布确认与重复，且错误头（x-death 语义）由 spring-rabbit 拼装而非 broker。

Spring AMQP：重试拦截器已换 Framework 7 重试 ​

spring-rabbit 4.0 的 RetryInterceptorBuilder.retryPolicy(...) 接收 Framework 7 的 org.springframework.core.retry.RetryPolicy（不再接收 spring-retry 的同名类）——升级 4.0 时所有自定义重试策略都要迁移。实验用 configureRetryPolicy(b -> b.maxRetries(2).delay(...)) 得到 3 次尝试。重试在消费端进行：每次重试都是一次完整重新投递，业务副作用必须幂等（回查重复消费与幂等）。

Spring Kafka：监听生命周期与 MANUAL ack 的偏移语义 ​

KafkaAdmin 自动创建主题（5 个 NewTopic）；KafkaTemplate.send(...).get() 同步确认；@KafkaListener 容器 start/stop 语义与 AMQP 相同（stop 期间消息留在分区，start 后从上次提交位点继续）。MANUAL ack 的实验揭示了它的真实粒度：ack 第 N 条会连同提交之前的 offset——实验先投 no-ack（不确认，重启容器后重新投递，at-least-once 成立），若同批次先投 no-ack 再投 acked 并 ack 后者，前者的 offset 也会被提交。批量消费场景里"逐条 ack"要配合 Acknowledgment.nack() 或分批策略，不能假设"没 ack 的那一条"独立存在。

Spring Kafka：错误处理的 seek、回退与 DLT ​

DefaultErrorHandler 组合 FixedBackOff 与 DeadLetterPublishingRecoverer：

原码java
	public DefaultErrorHandler(@Nullable ConsumerRecordRecoverer recoverer, BackOff backOff) {
		this(recoverer, backOff, null);
	}

	/**
	 * Construct an instance with the provided recoverer which will be called after the
	 * backOff returns STOP for a topic/partition/offset.
	 * @param recoverer the recoverer; if null, the default (logging) recoverer is used.
	 * @param backOff the {@link BackOff}.
	 * @param backOffHandler the {@link BackOffHandler}.
	 * @since 2.9
	 */
	public DefaultErrorHandler(@Nullable ConsumerRecordRecoverer recoverer, BackOff backOff,
			@Nullable BackOffHandler backOffHandler) {

		super(recoverer, backOff, backOffHandler, createFallback(backOff, recoverer));
	}

	private static CommonErrorHandler createFallback(BackOff backOff, @Nullable ConsumerRecordRecoverer recoverer) {
		return new FallbackBatchErrorHandler(backOff, recoverer);

失败记录按 backoff 重试（同一 offset seek 回来重投给监听器），耗尽后调用 recoverer 发布到死信主题。实验观察：poison 投递 3 次（首次+2 次重试）后进入 u48-t3-dlt。与 AMQP 的差异：Kafka 的"重试"不重新拉取消息，而是消费者 seek 回同一 offset——重试期间该分区消费停滞，大批量慢失败要用 nonBlockingRetry（多个重试主题）而不是阻塞 backoff。

Spring Kafka：事务的提交可见与中止不可见 ​

executeInTransaction 把模板操作包进生产者事务：

原码java
	public <T> @Nullable T executeInTransaction(OperationsCallback<K, V, T> callback) {
		Assert.notNull(callback, "'callback' cannot be null");
		Assert.state(this.transactional, "Producer factory does not support transactions");
		Thread currentThread = Thread.currentThread();
		Producer<K, V> producer = this.producers.get(currentThread);
		Assert.state(producer == null, "Nested calls to 'executeInTransaction' are not allowed");

		producer = this.producerFactory.createProducer(this.transactionIdPrefix);

		try {
			producer.beginTransaction();
		}
		catch (Exception e) {
			closeProducer(producer, false);
			throw e;
		}

		this.producers.put(currentThread, producer);
		try {
			T result = callback.doInOperations(this);
			try {
				producer.commitTransaction();
			}
			catch (Exception e) {

callback.doInOperations 执行后 commitTransaction，异常路径走 abort（跳过 commit 的 SkipAbortException）。实验用 read_committed 消费者验证了两件事：已提交事务的 2 条可见；事务内抛异常中止后，tx-aborted 对消费者不可见。事务边界是 producer 侧的（写多分区原子性），不等于业务事务；消费侧的"处理-提交"一致性要用容器事务或手工 ack 设计，回查 Spring 事务。

观测、排障与恢复 ​

症状先查什么本票依据
监听器启动报队列 404声明与容器启动的时序AMQP 观察一
消息"消失"没进 DLQbroker 是否支持 DLX；应用侧 recoverer 是否配置AMQP DLX 542 反例
Kafka 没有重复消费MANUAL ack 提交了同批之前 offsetMANUAL 语义
毒消息拖住分区DefaultErrorHandler 的 backoff 与 recovererkafkaErrorHandler 观察
事务消息"不见了"消费端 isolation.level 是否 read_committedkafkaTransaction 观察

容量与取舍 ​

监听容器并发数（concurrentConsumers / concurrency）决定单实例吞吐上限，但 AMQP 的预取（prefetch）与 Kafka 的 max.poll.records 才是内存与延迟的真正闸门；重试预算在消费端叠加业务幂等成本，阻塞式 backoff 会占住消费者线程；死信堆积需要清理策略。替身 broker 的边界：Qpid 覆盖 AMQP 0-9-1 的核心投递语义，但 rabbit 的 quorum 队列、DLX、消费者优先级等扩展不在其内——用替身验证的结论只覆盖 Spring 侧与协议核心。

条件变化与未运行边界 ​

未运行：RabbitMQ 本体（quorum 队列/DLX/发布确认对账）、Kafka 多 broker 与机架感知、Spring Pulsar（真实 broker 受阻）、并行容器与批量监听、性能容量。Kafka 为 4.1.2 KRaft 单机、Qpid 为 10.1.1 内存替身——替身结论不外推到 RabbitMQ 全部行为。教材"已讲清"不代表学习者掌握。

闭卷验收 ​

本页 12 道闭卷题收进了下方自测面板：先回忆，再展开参考答案与评分点，最后给三档自评。

来源与版本 ​

来源版本用途
RepublishMessageRecoverer.javav4.0.6应用侧死信（固定摘录）
DefaultErrorHandler.javav4.0.8seek 重试与 DLT（固定摘录）
KafkaTemplate.javav4.0.8事务边界（固定摘录）
Spring AMQP 4.0 参考、Spring Kafka 4.0 参考查询 2026-10-31官方配置与语义
Qpid Broker-J 10.1.110.1.1AMQP 替身 broker 边界

访问与运行日期：2026-10-31。源码片段为固定 tag 的连续行段，整文件 SHA-256 随每段元数据登记；运行版本以证据 JSON 为准。

工程学习台　阅读讲解，学后自主自测。记录保存在本机浏览器。