ReplyingKafkaTemplate源码分析

ReplyingKafkaTemplate是spring-kafka组件提供的一个用于实现请求响应模式的类，基础介绍可以参考文章https://blog.csdn.net/john1337/article/details/131363690，这里不再赘述这些细节，下面看下ReplyingKafkaTemplate是如何实现请求响应模式的。

	@Override
	public void afterPropertiesSet() {
		if (!this.schedulerSet && !this.schedulerInitialized) {
            // 初始化ThreadPoolTaskScheduler
			((ThreadPoolTaskScheduler) this.scheduler).initialize();
			this.schedulerInitialized = true;
		}
	}

	@Override
	public synchronized void start() {
		if (!this.running) {
			try {
				afterPropertiesSet();
			}
			catch (Exception e) {
				throw new KafkaException("Failed to initialize", e);
			}
            // 启动GenericMessageListenerContainer，用于后续接收响应数据
			this.replyContainer.start();
			this.running = true;
		}
	}

	public RequestReplyFuture<K, V, R> sendAndReceive(ProducerRecord<K, V> record, @Nullable Duration replyTimeout) {
		Assert.state(this.running, "Template has not been start()ed"); // NOSONAR (sync)
		// correlationId是本次消息的唯一标识，实际上就是一个random 的uuid组成，这部分代码就不再列出，感兴趣的可以查看ReplyingKafkaTemplate类defaultCorrelationIdStrategy方法
        CorrelationKey correlationId = this.correlationStrategy.apply(record);
		Assert.notNull(correlationId, "the created 'correlationId' cannot be null");
		Headers headers = record.headers();
		boolean hasReplyTopic = headers.lastHeader(KafkaHeaders.REPLY_TOPIC) != null;
		if (!hasReplyTopic && this.replyTopic != null) {
			headers.add(new RecordHeader(this.replyTopicHeaderName, this.replyTopic));
			if (this.replyPartition != null) {
				headers.add(new RecordHeader(this.replyPartitionHeaderName, this.replyPartition));
			}
		}
		headers.add(new RecordHeader(this.correlationHeaderName, correlationId.getCorrelationId()));
		this.logger.debug(() -> "Sending: " + record + WITH_CORRELATION_ID + correlationId);
        // 创建RequestReplyFuture，并存入futures中
		RequestReplyFuture<K, V, R> future = new RequestReplyFuture<>();
		this.futures.put(correlationId, future);
		try {
			future.setSendFuture(send(record));
		}
		catch (Exception e) {
			this.futures.remove(correlationId);
			throw new KafkaException("Send failed", e);
		}
        // 启动定时检测，指定时间结束后及时没收到反馈，也结束等待并返回KafkaReplyTimeoutException超时异常
		scheduleTimeout(record, correlationId, replyTimeout == null ? this.defaultReplyTimeout : replyTimeout);
		return future;
	}

	private void scheduleTimeout(ProducerRecord<K, V> record, CorrelationKey correlationId, Duration replyTimeout) {
        // 超时检测部分代码
		this.scheduler.schedule(() -> {
			RequestReplyFuture<K, V, R> removed = this.futures.remove(correlationId);
			if (removed != null) {
				this.logger.warn(() -> "Reply timed out for: " + record + WITH_CORRELATION_ID + correlationId);
				if (!handleTimeout(correlationId, removed)) {
					removed.setException(new KafkaReplyTimeoutException("Reply timed out"));
				}
			}
		}, Instant.now().plus(replyTimeout));
	}

	public void onMessage(List<ConsumerRecord<K, R>> data) {
		data.forEach(record -> {
            // 获取kafka_correlationId请求头
			Header correlationHeader = record.headers().lastHeader(this.correlationHeaderName);
			CorrelationKey correlationId = null;
			if (correlationHeader != null) {
				correlationId = new CorrelationKey(correlationHeader.value());
			}
			if (correlationId == null) {
				this.logger.error(() -> "No correlationId found in reply: " + record
						+ " - to use request/reply semantics, the responding server must return the correlation id "
						+ " in the '" + this.correlationHeaderName + "' header");
			}
			else {
                // 收到对应反馈数据，这里需要判断是否该数据在来到之前已被超时检测删除
				RequestReplyFuture<K, V, R> future = this.futures.remove(correlationId);
				CorrelationKey correlationKey = correlationId;
				if (future == null) {
                     // 数据已被超时检测机制删除，记录日志
					logLateArrival(record, correlationId);
				}
				else {
					this.logger.debug(() -> "Received: " + record + WITH_CORRELATION_ID + correlationKey);
                     // 将反馈返回
					future.set(record);
				}
			}
		});
	}