1.C# 连接Kafka知识分享
前些天公司的Boss突然下达一个命令，消息中间件要用Kafka，既然领导都决定了用就用呗。那就网上百度一下去Kafka如何安装啊，Kafka用代码如何连接操作。在安装和使用过过程中遇到了一些坎坷的事情，最总还是解决了。
我所在部门使用C#编程语言，所以连接Kafka用C#语言去实现，可能朋友们会说那不是很简单吗？百度一下网上一大堆。百度是一大堆但未必是你想要的，网上找了好多篇都是基于Java语言编写的，C#的也有，但是没Java资料丰富。
https://gitee.com/autumn_2/MQExtend.Core.git 基于MQ提供的Sdk。二次封装后支持对ActiveMQ、Kafak相关操作（本人也是一个小白，写的东西也是半桶水。但希望对大家有帮助）
2.Confluent kafka
在选择Kafka类库之前看了https://blog.csdn.net/xinlingjun2007/article/details/80295332 这篇博客，所以就选择了Confluent kafka 类库了
2.1 消息发送
kafka 中的auto.create.topics.enable默认为false的，所以在发送消息给Topics之前，确保Topics在Kafka里要存在。这个需要注意一下。
public void PushMessage()
{
var config = new ProducerConfig()
{
BootstrapServers = "localhost:9092",
Acks = Acks.Leader
};
//message<key,value> 这个key目前没用，做消息指定分区投放有用的；我们直接用null
using(var producer = new ProducerBuilder<Null, string>(config).Build())
{
producer.Produce("TopicName", new Message<Null, string>()
{
Value = "需要发送的消息内容"
}, (result) =>
{
WriteLog(!result.Error.IsError ? $"Delivered message to {result.TopicPartitionOffset}" : $"Delivery Error: {result.Error.Reason}");
});
Console.WriteLine("消息发送成功");
}
}
2.2 消息消费
session.timeout.ms
如果consumer在这段时间内没有发送心跳信息，则它会被认为挂掉了。默认3秒。
auto.offset.reset
消费者在读取一个没有偏移量的分区或者偏移量无效的情况下，如何处理。默认值是latest。
earliest：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
latest：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
none：各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
enable .auto.commit
默认值true，表明消费者是否自动提交偏移。为了尽量避免重复数据和数据丢失，可以改为false，自行控制何时提交
/// <summary>
/// 消息订阅
/// </summary>
/// <param name="subscribe"></param>
public void Subscribe(string queueName, Action<IMessageContent> action)
{
var config = new ConsumerConfig()
{
BootstrapServers = "",
GroupId = "gms_20200327_group",
AutoOffsetReset = AutoOffsetReset.Earliest,
EnableAutoCommit = false
};
//如果Kafka配置了安全认证（我这里只是写案例，本地配置了sasl安全认证）就加这块代码
if (!string.IsNullOrEmpty(this.UserName) && !string.IsNullOrEmpty(this.Password))
{
config.SecurityProtocol = SecurityProtocol.SaslPlaintext;
config.SaslMechanism = SaslMechanism.Plain;
config.SaslUsername = this.UserName;
config.SaslPassword = this.Password;
}
using (var consumer = new ConsumerBuilder<Ignore, string>(config).Build())
{
//订阅topicName
consumer.Subscribe(queueName);
CancellationTokenSource cts = new CancellationTokenSource();
Console.CancelKeyPress += (sender, e) =>
{
//prevent the process from terminating.
e.Cancel = true;
cts.Cancel();
};
//是否消费成功
bool isOK = false;
//result
ConsumeResult<Ignore, string> consumeResult = null;
try
{
while (true)
{
isOK = false;
try
{
//consumer.Assign(new TopicPartitionOffset(queueName, 0, Offset.Beginning));
consumeResult = consumer.Consume(cts.Token);
if (consumeResult.IsPartitionEOF)
{
WriteLog($"Reached end of topic {consumeResult.Topic}, partition {consumeResult.Partition}, offset {consumeResult.Offset}.");
continue;
}
//接收到的消息记录Log
WriteLog($"Received message at {consumeResult.TopicPartitionOffset}: {consumeResult.Value}");
//消息消费
action?.Invoke(new KafkaMessageContent(consumeResult.Value, consumeResult.Key?.ToString()));
//消费成功
isOK = true;
//提交方法向Kafka集群发送一个“提交偏移量”请求，并同步等待响应。
//与消费者能够消费消息的速度相比，这是非常慢的。
//一个高性能的应用程序通常会相对不频繁地提交偏移量，并且在失败的情况下被设计来处理重复的消息
consumer.Commit(consumeResult);
//消费成功Log记录
WriteLog($"Consumed message '{consumeResult.Value}' at: '{consumeResult.TopicPartitionOffset}'.");
}
catch (ConsumeException e)
{
isOK = false;
WriteError($"Error occured: {e.Error.Reason}");
}
catch (Exception ex)
{
isOK = false;
WriteError($"Error occured: {ex.StackTrace}");
}
//消费失败后置处理
if (!isOK && consumeResult != null)
{
//消费失败代码逻辑处理
ErrorHandler(consumer, consumeResult);
}
}
}
catch (OperationCanceledException e)
{
WriteException(e);
// Ensure the consumer leaves the group cleanly and final offsets are committed.
consumer.Close();
}
}
}
注意：这里有一点点需要注意下consumer.Commit(consumeResult)。我先列举一个例子如果一个分区里面有10条消息
| 1 |
2 |
3（消费失败） |
4 |
5 |
6 |
7 |
8 |
9 |
10 |
- 如果3这条消息消费失败，那么就会被catch捕获，代码没执行到consumer.Commit() 提交偏移量这段代码。
- 因为异常被catch了，所以消费者继续poll消息，获取到4这条消息。4这条消息成功了，就会执行consumer.Commit() 提交偏移量。（注意：这里就提交了最新的偏移量了），换句话说如果3这条消费失败，不去做一些额外的处理3这条消息就消费不到了（注意：不人工干预是消费不到的）。
- 我这边处理逻辑是将消费失败的消息将其转发到DLQ队列，就是创建一个Topics的同时在创建一个DLQ.Topics。这个DLQ.Topics专门用来存放 Topics消费失败的消息。处理代码如下：
/// <summary>
/// 消费异常处理
/// </summary>
/// <param name="consumer">消费者</param>
/// <param name="consumeResult">消息</param>
private void ErrorHandler(IConsumer<Ignore, string> consumer, ConsumeResult<Ignore, string> consumeResult)
{
if (consumeResult != null && consumer != null)
{
string queueName = consumeResult.Topic;
WriteLog($"Consumed '{queueName}' message fail '{consumeResult.Value}' at: '{consumeResult.TopicPartitionOffset}'.");
//消费失败，并且需要不需要转发到DLQ队列中，所以我们这里需要把(Offset-1)
if (!this.SubscribeConfig.TransformToDLQ || queueName.StartsWith("DLQ.", StringComparison.OrdinalIgnoreCase))
{
//偏移量往回拉一位，尝试6次操作。如果执行失败，确保消息不遗漏直接停止消费。
OffsetBack(consumer, consumeResult);
return;
}
//需要转发到DLQ队列中
string transformTopics = "DLQ." + queueName;
WriteLog($"消息开始转发到{transformTopics}队列");
KafkaProducerConfig config = new KafkaProducerConfig(ServerConfig.BrokerUri,
ServerConfig.UserName, ServerConfig.Password, transformTopics);
try
{
//将消息转发到死信队列
using (IProducerChannel producer = new KafkaProducer(config))
{
producer.Producer(consumeResult.Value?.ToString());
}
//提交偏移量
consumer.Commit(consumeResult);
WriteLog($"消息转发到{transformTopics}队列成功");
}
catch (Exception ex)
{
WriteError($"消息转发到{transformTopics}队列失败。Error occured: {ex.StackTrace}");
//偏移量往回拉一位，尝试6次操作。如果执行失败，确保消息不遗漏直接停止消费。
OffsetBack(consumer, consumeResult);
}
}
}
/// <summary>
/// 把Offset偏移量往回拉一位
/// </summary>
/// <param name="consumer"></param>
/// <param name="consumeResult"></param>
/// <param name="tryTimes">默认执行6次</param>
private void OffsetBack(IConsumer<Ignore, string> consumer, ConsumeResult<Ignore, string> consumeResult, int tryTimes = 6)
{
int count = tryTimes;
string queueName = consumeResult.Topic;
while (count > 0)
{
WriteLog($"消息消费失败，执行偏移量Offset-1操作");
try
{
//消费失败，重置一下最新偏移量
consumer.Assign(new TopicPartitionOffset(queueName, consumeResult.Partition, consumeResult.Offset));
WriteLog($"偏移量重置成功{consumeResult.Offset}");
count--;
return;
}
catch (Exception ex)
{
WriteLog($"消息消费失败，执行偏移量Offset-1操作失败。Error occured: {ex.StackTrace}");
//尝试重置偏移量次数到了最大次数，直接抛出异常。停止消费
if (count == 0)
{
WriteError($"消息消费失败，执行偏移量Offset-1操作失败次数已达到${tryTimes}，消费者停止消费");
//抛出这个异常，会引发Subscribe()到catch代码块。catch会停止消费
throw new OperationCanceledException($"消息消费失败，执行偏移量Offset-1操作失败次数已达到${tryTimes}，消费者停止消费");
}
//停止3s在重新重置偏移量
Thread.Sleep(3000);
}
}
}
- 以上的消费异常处理只是本人的观点，可能有更好的处理方案，输入在下方一起交交流。
