Spring Cloud Stream 中的事务 [英] Transaction in Spring cloud Stream
问题描述
问题:我正在尝试逐行读取一个大文件并将消息放入 RabbitMQ.我想在文件末尾提交rabbitMQ.如果文件中有任何记录是坏的,那么我想撤消发布到队列中的消息.
Problem: I am trying to read a big file line by line and putting the message in a RabbitMQ. I want to commit to rabbitMQ at the end of the file. If any record in the file is bad, then I want to revoke the messages published to the queue.
技术:弹簧靴,春云流,兔MQ
Technologies: Spring boot, Spring cloud stream, RabbitMQ
你能帮我实现这个过渡的东西吗?我知道如何使用 Spring Cloud 流读取文件并发布到队列.
Could you please help me in implementing this transition stuff. I know how to read a file and publish to a queue using spring cloud stream.
@Transactional
public void sendToQueue(List<Data> dataList) {
for(Data data:dataList)
{
this.output.send(MessageBuilder.withPayload(data).build());
counter++; // I can see message getting published in the queue though management plugin
}
LOGGER.debug("message sent to Q2");
}
这是我的配置:
spring:
cloud:
stream:
bindings:
# Q1 input channel
tpi_q1_input:
destination: TPI_Q1
binder: local_rabbit
content-type: application/json
group: TPIService
# Q2 output channel
tpi_q2_output:
destination: TPI_Q2
binder: local_rabbit
content-type: application/json
group: TPIService
# Q2 input channel
tpi_q2_input:
destination: TPI_Q2
binder: local_rabbit
content-type: application/json
group: TPIService
binders:
local_rabbit:
type: rabbit
environment:
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
virtual-host: /
rabbit:
bindings:
tpi_q2_output:
producer:
#autoBindDlq: true
transacted: true
#batchingEnabled: true
tpi_q2_input:
consumer:
acknowledgeMode: AUTO
#autoBindDlq: true
#recoveryInterval: 5000
transacted: true
spring.cloud.stream.default-binder: local_rabbit
Java 配置
@EnableTransactionManagement
public class QueueConfig {
@Bean
public RabbitTransactionManager transactionManager(ConnectionFactory cf) {
return new RabbitTransactionManager(cf);
}
}
接收方
@StreamListener(JmsQueueConstants.QUEUE_2_INPUT)
@Transactional
public void receiveMesssage(Data data) {
logger.info("Message Received in Q2:");
}
推荐答案
配置生产者使用事务
...producer.transacted=true
在事务范围内发布消息(使用 RabbitTransactionManager).
Publish the messages within the scope of a transaction (using the RabbitTransactionManager).
对 #2 使用普通的 Spring 事务机制(@Transacted
注释或 TransactionTemplate
).
Use normal Spring transaction mechanisms for #2 (@Transacted
annotation or a TransactionTemplate
).
如果正常退出事务会提交,如果抛出异常则回滚.
The transaction will commit if you exit normally, or roll back if you throw an exception.
编辑
示例:
@SpringBootApplication
@EnableBinding(Source.class)
@EnableTransactionManagement
public class So50372319Application {
public static void main(String[] args) {
SpringApplication.run(So50372319Application.class, args).close();
}
@Bean
public ApplicationRunner runner(MessageChannel output, RabbitTemplate template, AmqpAdmin admin,
TransactionalSender sender) {
admin.deleteQueue("so50372319.group");
admin.declareQueue(new Queue("so50372319.group"));
admin.declareBinding(new Binding("so50372319.group", DestinationType.QUEUE, "output", "#", null));
return args -> {
sender.send("foo", "bar");
System.out.println("Received: " + new String(template.receive("so50372319.group", 10_000).getBody()));
System.out.println("Received: " + new String(template.receive("so50372319.group", 10_000).getBody()));
try {
sender.send("baz", "qux");
}
catch (RuntimeException e) {
System.out.println(e.getMessage());
}
System.out.println("Received: " + template.receive("so50372319.group", 3_000));
};
}
@Bean
public RabbitTransactionManager transactionManager(ConnectionFactory cf) {
return new RabbitTransactionManager(cf);
}
}
@Component
class TransactionalSender {
private final MessageChannel output;
public TransactionalSender(MessageChannel output) {
this.output = output;
}
@Transactional
public void send(String... data) {
for (String datum : data) {
this.output.send(new GenericMessage<>(datum));
if ("qux".equals(datum)) {
throw new RuntimeException("fail");
}
}
}
}
和
spring.cloud.stream.bindings.output.destination=output
spring.cloud.stream.rabbit.bindings.output.producer.transacted=true
和
Received: foo
Received: bar
fail
Received: null
这篇关于Spring Cloud Stream 中的事务的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持IT屋!