.net core 下使用 Kafka 生产者批量发送给消息处理，使用事务(四)

生产者批量发送消息，使用事务，要么全部失败要么全部成功

重要 说明 事物id必须要设置

producerConfig.TransactionalId =Guid.NewGuid().ToString();//必须设置事物id

 1   /// <summary>
 2         /// 批量发送
 3         /// </summary>
 4         /// <param name="msglist"></param>
 5 
 6         public void BatchSends(List<string> msglist)
 7         {
 8             ProducerConfig producerConfig = this.CreateProducerConnection();
 9             producerConfig.MessageTimeoutMs = 5000;//失败重试时间
10             producerConfig.EnableIdempotence = true;///幂等性:如果生产者发送失败不重复发消息失败重试
11             producerConfig.TransactionalId =Guid.NewGuid().ToString();//必须设置事物id
12             var builder = new ProducerBuilder<string, string>(producerConfig);
13             builder.SetDefaultPartitioner(RoundRobinPartitioner);//委托:调用负载均衡方法
14             using (var producer = builder.Build())
15             {
16                 // 1、初始化事务
17                 producer.InitTransactions(TimeSpan.FromSeconds(60));
18                 try
19                 {
20                     // 2、开发事务
21                     producer.BeginTransaction();
22                     foreach (var item in msglist)
23                     {
24                         //这里就是测试
25                        // var OrderJson = JsonConvert.SerializeObject(item);
26                        // var dr = producer.ProduceAsync("order-create", new Message<string, string> { Key = "order", Value = OrderJson }).GetAwaiter().GetResult();
27                         var dr = producer.ProduceAsync(KafkaTopic.Topic, new Message<string, string> { Key = "order", Value = item }).GetAwaiter().GetResult();
28                         _logger.LogInformation("发送事件 {0} 到 {1} 成功", dr.Value, dr.TopicPartitionOffset);
29                     }
30                     // 3、提交事务
31                     producer.CommitTransaction();
32                 }
33                 catch (ProduceException<string, string> ex)
34                 {
35                     _logger.LogError(ex, "发送事件到 {0} 失败，原因 {1} ", "order", ex.Error.Reason);
36                     // 4、关闭事务
37                     producer.AbortTransaction();
38                 }
39             }
40         }

webapi 项目调用

 1         /// <summary>
 2         /// 批量发送消息
 3         /// </summary>
 4         /// <param name="msglist"></param>
 5         /// <returns></returns>
 6         [HttpPost("BatchSend")]
 7         public string BatchSend(List<string> msglist) {
 8 
 9             this._kafKaSevice.BatchSends(msglist);
10             return "";
11         }    

消息消费还是用上一篇的代码就行不需要更改 还是粘出来吧

 1   /// <summary>
 2         /// 消费消息 redis 存储 offset 偏移量
 3         /// </summary>
 4         public void Reveice()
 5         {
 6             ConsumerConfig consumerConfig = this.CreateConsumerConnection();
 7             //AutoOffsetReset.Latest 这里是kafka 停止重启 从最新产生的数据开始消费
 8             //AutoOffsetReset.Earliest 这里是kafka 停止重启 从第0个索引开始消费
 9             //earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
10             //latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
11             consumerConfig.AutoOffsetReset = AutoOffsetReset.Earliest;
12             consumerConfig.GroupId = "order";//分组为order组
13             //consumerConfig.EnableAutoCommit = true;
14             //EnableAutoCommit = true//自动确认 如果设置false 会被重复消费
15             //缺陷 当业务逻辑 失败时候 可以自动确认 导致 消息丢失
16             //所以 要手动确认 但是如果手动确认时候 kafka服务宕机 重启时候会导致重复消费
17             //此时就需要 偏移量+1 重置 确保偏移量 是最新的
18             var builder = new ConsumerBuilder<string, string>(consumerConfig);
19             string redisValue = this._distributedCache.GetString(KafkaTopic.Topic);
20             int offset = 0;
21             if (!string.IsNullOrEmpty(redisValue))
22             {
23                 offset = int.Parse(redisValue);
24             }
25             this._distributedCache.SetString(KafkaTopic.Topic, (offset + 1).ToString());
26             using (var consumer = builder.Build())
27             {
28                 //offset重置
29                 //consumer.Assign(new TopicPartition(KafkaTopic.Topic, new Partition(1)));
30                 consumer.Assign(new TopicPartitionOffset(KafkaTopic.Topic, new Partition(0), offset + 1));
31                 while (true)
32                 {
33                     // 订阅 生产者的主题
34                     consumer.Subscribe(KafkaTopic.Topic);
35                     // 消息 获取
36                     var result = consumer.Consume();
37                     // 获取偏移量
38                     _logger.LogInformation($"订单消息偏移量：Offset:{result.Offset}");
39                     // 把kafka队列中偏移量存起来。redis mysql
40                     // redis存储 kafka队列的偏移量
41                     _distributedCache.SetString("create-order", result.Offset.Value.ToString());
42                     string key = result.Key;
43                     string value = result.Value;
44                     //使用了redis 缓存 offset 重置 就不需要手动确认了
45                     //consumer.Commit(result);
46                     //consumer.StoreOffset(result);
47                     this._logger.LogError($" 收到消息了:{value} / offset:{result.Offset}/Partition: {result.Partition} ");
48                 }
49             }
50         }

测试