我想使用一些回调 api 为我的 rabbitmq 生产者找出失败的消息。我已经用 [{rabbit, [{vm_memory_high_watermark, 0.001}]}] 配置了 rabbitmq。并尝试推送大量消息,但所有消息都被接受,并且稍后会出现 TimeoutException 并且消息没有发送到 Queue enter code here
,请告诉我如何捕获它。
发送消息代码:
// #create-sink - producer
final Sink<ByteString, CompletionStage<Done>> amqpSink =
AmqpSink.createSimple(
AmqpSinkSettings.create(connectionProvider)
.withRoutingKey(AkkaConstants.queueName)
.withDeclaration(queueDeclaration));
// #run-sink
//final List<String> input = Arrays.asList("one", "two", "three", "four", "five");
//Source.from(input).map(ByteString::fromString).runWith(amqpSink, materializer);
String filePath = "D:\\subrata\\code\\akkaAmqpTest-master\\akkaAmqpTest-master\\logs2\\dummy.txt";
Path path = Paths.get(filePath);
// List containing 78198 individual message
List<String> contents = Files.readAllLines(path);
System.out.println("********** file reading done ....");
int times = 5;
// Send 78198*times message to Queue [From console i can see 400000 number of messages being sent]
for(int i=0;i<times;i++) {
Source.from(contents).map(ByteString::fromString).runWith(amqpSink, materializer);
}
System.out.println("************* sending to queue is done");