Flink 与rabbitmq集成 并开启checkpoint

如果不开启checkpoint机制，flink job 在运行时如果遇到异常整个job就会停止。

如果开启了checkpoint机制，就会根据恢复点进行数据重试，这个是一个非常复杂的机制，需要单独的文章进行解析。

所以开启checkpoint是必然要做的配置。

在与rabbitmq集成时有个点必须要注意，就是mq 发送消息时候，必须要带上CorrelationId。我们看一下flink的官方文档。

env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

env.getCheckpointConfig().setCheckpointInterval(10000);

DataStream dataStreamSource = env.addSource(new RMQSource(connectionConfig,

"kgraph",true,new SimpleStringSchema()));

dataStreamSource.print();

上面是构造RMQSource(…)的参数，如下

queueName: The RabbitMQ queue name.

usesCorrelationId: true when correlation ids should be used, false otherwise (default is false).

deserializationScehma: Deserialization schema to turn messages into Java objects.

根据参数不同，有如下三种模式

Exactly-once (when checkpointed) with RabbitMQ transactions and messages with unique correlation IDs.

At-least-once (when checkpointed) with RabbitMQ transactions but no deduplication mechanism (correlation id is not set).

No strong delivery guarantees (without checkpointing) with RabbitMQ auto-commit mode.

那么 开启 exactly-once 确保消费一次的特性，就必须在传递 mq消息的时候带上 correlationId。

correlation Id 是 mq 消息的一个基本属性，可以用来标识消息的唯一id，通常是mq实现rpc调用时使用，flink 利用其唯一id的特性来做 exactly once的消费。所以在发送mq消息时 加上 correlation_id 的properties 就不会有问题了。

如果使用 spring 结合 rabbitmq 作为客户端，需要对 correlationId 做一个特别的处理

就是需要自己手动设置correaltionId， rabbittemplate 没有自动的封装这个属性，convertAndSend这个方法非常让人confuse，

里面支持传入correlationData字段，但是这个是写入到消息头的，而不是correlation_id，flink那边永远无法读取到。

public void sendMsg(KgraphUpdateMessage kgraphUpdateMessage)

{

CorrelationData correlationId =new CorrelationData(UUID.randomUUID().toString());

ObjectMapper jsonReader =new ObjectMapper();

try

    {

MessageProperties properties =new MessageProperties();

properties.setCorrelationId(correlationId.getId().getBytes());

Message message =new Message(jsonReader.writeValueAsBytes(kgraphUpdateMessage), properties);

rabbitTemplate.convertAndSend(KgraphMqConfig.KGRAPH_EXCHANGE, KgraphMqConfig.KGRAPH_TOPIC_EVENT, message,correlationId);

}catch (JsonProcessingException e)

{

e.printStackTrace();

}

}