我已经配置了一个spring云流kafka应用程序来使用事务(github上提供了完整的源代码):
spring:
application:
name: message-relay-service
cloud:
stream:
kafka:
binder:
transaction:
transaction-id-prefix: message-relay-tx-
producer:
configuration:
retries: 1
acks: all
key:
serializer: org.apache.kafka.common.serialization.StringSerializer
bindings:
output:
destination: transfer
contentType: application/*+avro
schema-registry-client:
endpoint: http://localhost:8081
schema:
avro:
subjectNamingStrategy: org.springframework.cloud.stream.schema.avro.QualifiedSubjectNamingStrategy
datasource:
url: jdbc:h2:tcp://localhost:9090/mem:mydb
driver-class-name: org.h2.Driver
username: sa
password:
jpa:
hibernate:
ddl-auto: create
database-platform: org.hibernate.dialect.H2Dialect
server:
port: 8085
此应用程序有一个计划任务,定期检查要使用 @Scheduled
任务。此方法的注解为 @Transactional
主类定义 @EnableTransactionManagement
.
但是,在调试代码时,我意识到kafkatransactionmanager没有被执行,也就是说,没有kafka事务。有什么问题吗?
@EnableTransactionManagement
@EnableBinding(Source::class)
@EnableScheduling
@SpringBootApplication
class MessageRelayServiceApplication
fun main(args: Array<String>) {
runApplication<MessageRelayServiceApplication>(*args)
}
---
@Component
class MessageRelay(private val outboxService: OutboxService,
private val source: Source) {
@Transactional
@Scheduled(fixedDelay = 10000)
fun checkOutbox() {
val pending = outboxService.getPending()
pending.forEach {
val message = MessageBuilder.withPayload(it.payload)
.setHeader(KafkaHeaders.MESSAGE_KEY, it.messageKey)
.setHeader(MessageHeaders.CONTENT_TYPE, it.contentType)
.build()
source.output().send(message)
outboxService.markAsProcessed(it.id)
}
}
}
1条答案
按热度按时间p8ekf7hl1#
我不明白
@EnableTransactionManagement
在account-service
,仅在message-relay-service
.在任何情况下,您的方案目前都不受支持;事务绑定器是为这样的处理器设计的:使用者启动事务,使用者线程上发送的任何记录都参与该事务,使用者向事务发送偏移量,然后提交事务。
它不是专为生产者绑定;请针对活页夹打开github问题,因为它应该得到支持。
我不知道为什么你没有看到一个交易开始,但是,即使是这样,问题是
@Transactional
将使用boot的自动配置ktm(和producer-factory),绑定使用不同的producer-factory(您配置的那个)。即使交易正在进行中,制作人也不会参与其中。