.NET Core下使用Kafka的方法步骤

 更新时间:2020-11-19 04:07:33   作者:佚名   我要评论(0)

安装
CentOS安装 kafka
Kafka : http://kafka.apache.org/downloads
ZooLeeper : https://zookeeper.apache.org/releases.html
下载并解压


# 下载,并解压
$ wget

安装

CentOS安装 kafka

Kafka : http://kafka.apache.org/downloads

ZooLeeper : https://zookeeper.apache.org/releases.html

下载并解压

# 下载,并解压
$ wget https://archive.apache.org/dist/kafka/2.1.1/kafka_2.12-2.1.1.tgz
$ tar -zxvf kafka_2.12-2.1.1.tgz
$ mv kafka_2.12-2.1.1.tgz /data/kafka

# 下载 zookeeper,解压
$ wget https://mirror.bit.edu.cn/apache/zookeeper/zookeeper-3.5.8/apache-zookeeper-3.5.8-bin.tar.gz
$ tar -zxvf apache-zookeeper-3.5.8-bin.tar.gz
$ mv apache-zookeeper-3.5.8-bin /data/zookeeper

启动 ZooKeeper

# 复制配置模版
$ cd /data/kafka/conf
$ cp zoo_sample.cfg zoo.cfg

# 看看配置需不需要改
$ vim zoo.cfg

# 命令
$ ./bin/zkServer.sh start  # 启动
$ ./bin/zkServer.sh status  # 状态
$ ./bin/zkServer.sh stop   # 停止
$ ./bin/zkServer.sh restart # 重启

# 使用客户端测试
$ ./bin/zkCli.sh -server localhost:2181
$ quit

启动 Kafka

# 备份配置
$ cd /data/kafka
$ cp config/server.properties config/server.properties_copy

# 修改配置
$ vim /data/kafka/config/server.properties

# 集群配置下,每个 broker 的 id 是必须不同的
# broker.id=0

# 监听地址设置(内网)
# listeners=PLAINTEXT://ip:9092

# 对外提供服务的IP、端口
# advertised.listeners=PLAINTEXT://106.75.84.97:9092

# 修改每个topic的默认分区参数num.partitions,默认是1,具体合适的取值需要根据服务器配置进程确定,UCloud.ukafka = 3
# num.partitions=3

# zookeeper 配置
# zookeeper.connect=localhost:2181

# 通过配置启动 kafka
$ ./bin/kafka-server-start.sh config/server.properties&

# 状态查看
$ ps -ef|grep kafka
$ jps

docker下安装Kafka

docker pull wurstmeister/zookeeper
docker run -d --name zookeeper -p 2181:2181 wurstmeister/zookeeper
docker pull wurstmeister/kafka
docker run -d --name kafka --publish 9092:9092 --link zookeeper --env KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 --env KAFKA_ADVERTISED_HOST_NAME=192.168.1.111 --env KAFKA_ADVERTISED_PORT=9092 wurstmeister/kafka

介绍

  • Broker:消息中间件处理节点,一个Kafka节点就是一个broker,多个broker可以组成一个Kafka集群。
  • Topic:一类消息,例如page view日志、click日志等都可以以topic的形式存在,Kafka集群能够同时负责多个topic的分发。
  • Partition:topic物理上的分组,一个topic可以分为多个partition,每个partition是一个有序的队列。
  • Segment:partition物理上由多个segment组成,下面2.2和2.3有详细说明。
  • offset:每个partition都由一系列有序的、不可变的消息组成,这些消息被连续的追加到partition中。partition中的每个消息都有一个连续的序列号叫做offset,用于partition唯一标识一条消息。

kafka partition 和 consumer 数目关系

  • 如果consumer比partition多是浪费,因为kafka的设计是在一个partition上是不允许并发的,所以consumer数不要大于partition数 。
  • 如果consumer比partition少,一个consumer会对应于多个partitions,这里主要合理分配consumer数和partition数,否则会导致partition里面的数据被取的不均匀 。最好partiton数目是consumer数目的整数倍,所以partition数目很重要,比如取24,就很容易设定consumer数目 。
  • 如果consumer从多个partition读到数据,不保证数据间的顺序性,kafka只保证在一个partition上数据是有序的,但多个partition,根据你读的顺序会有不同
  • 增减consumer,broker,partition会导致rebalance,所以rebalance后consumer对应的partition会发生变化 快速开始

在 .NET Core 项目中安装组件

Install-Package Confluent.Kafka

开源地址: https://github.com/confluentinc/confluent-kafka-dotnet

添加 IKafkaService 服务接口

public interface IKafkaService
{
  /// <summary>
  /// 发送消息至指定主题
  /// </summary>
  /// <typeparam name="TMessage"></typeparam>
  /// <param name="topicName"></param>
  /// <param name="message"></param>
  /// <returns></returns>
  Task PublishAsync<TMessage>(string topicName, TMessage message) where TMessage : class;

  /// <summary>
  /// 从指定主题订阅消息
  /// </summary>
  /// <typeparam name="TMessage"></typeparam>
  /// <param name="topics"></param>
  /// <param name="messageFunc"></param>
  /// <param name="cancellationToken"></param>
  /// <returns></returns>
  Task SubscribeAsync<TMessage>(IEnumerable<string> topics, Action<TMessage> messageFunc, CancellationToken cancellationToken) where TMessage : class;
}

实现 IKafkaService

public class KafkaService : IKafkaService
{
  public async Task PublishAsync<TMessage>(string topicName, TMessage message) where TMessage : class
  {
    var config = new ProducerConfig
    {
      BootstrapServers = "127.0.0.1:9092"
    };
    using var producer = new ProducerBuilder<string, string>(config).Build();
    await producer.ProduceAsync(topicName, new Message<string, string>
    {
      Key = Guid.NewGuid().ToString(),
      Value = message.SerializeToJson()
    });
  }

  public async Task SubscribeAsync<TMessage>(IEnumerable<string> topics, Action<TMessage> messageFunc, CancellationToken cancellationToken) where TMessage : class
  {
    var config = new ConsumerConfig
    {
      BootstrapServers = "127.0.0.1:9092",
      GroupId = "crow-consumer",
      EnableAutoCommit = false,
      StatisticsIntervalMs = 5000,
      SessionTimeoutMs = 6000,
      AutoOffsetReset = AutoOffsetReset.Earliest,
      EnablePartitionEof = true
    };
    //const int commitPeriod = 5;
    using var consumer = new ConsumerBuilder<Ignore, string>(config)
               .SetErrorHandler((_, e) =>
               {
                 Console.WriteLine($"Error: {e.Reason}");
               })
               .SetStatisticsHandler((_, json) =>
               {
                 Console.WriteLine($" - {DateTime.Now:yyyy-MM-dd HH:mm:ss} > 消息监听中..");
               })
               .SetPartitionsAssignedHandler((c, partitions) =>
               {
                 string partitionsStr = string.Join(", ", partitions);
                 Console.WriteLine($" - 分配的 kafka 分区: {partitionsStr}");
               })
               .SetPartitionsRevokedHandler((c, partitions) =>
               {
                 string partitionsStr = string.Join(", ", partitions);
                 Console.WriteLine($" - 回收了 kafka 的分区: {partitionsStr}");
               })
               .Build();
    consumer.Subscribe(topics);
    try
    {
      while (true)
      {
        try
        {
          var consumeResult = consumer.Consume(cancellationToken);
          Console.WriteLine($"Consumed message '{consumeResult.Message?.Value}' at: '{consumeResult?.TopicPartitionOffset}'.");
          if (consumeResult.IsPartitionEOF)
          {
            Console.WriteLine($" - {DateTime.Now:yyyy-MM-dd HH:mm:ss} 已经到底了:{consumeResult.Topic}, partition {consumeResult.Partition}, offset {consumeResult.Offset}.");
            continue;
          }
          TMessage messageResult = null;
          try
          {
            messageResult = JsonConvert.DeserializeObject<TMessage>(consumeResult.Message.Value);
          }
          catch (Exception ex)
          {
            var errorMessage = $" - {DateTime.Now:yyyy-MM-dd HH:mm:ss}【Exception 消息反序列化失败,Value:{consumeResult.Message.Value}】 :{ex.StackTrace?.ToString()}";
            Console.WriteLine(errorMessage);
            messageResult = null;
          }
          if (messageResult != null/* && consumeResult.Offset % commitPeriod == 0*/)
          {
            messageFunc(messageResult);
            try
            {
              consumer.Commit(consumeResult);
            }
            catch (KafkaException e)
            {
              Console.WriteLine(e.Message);
            }
          }
        }
        catch (ConsumeException e)
        {
          Console.WriteLine($"Consume error: {e.Error.Reason}");
        }
      }
    }
    catch (OperationCanceledException)
    {
      Console.WriteLine("Closing consumer.");
      consumer.Close();
    }
    await Task.CompletedTask;
  }
}

注入 IKafkaService ,在需要使用的地方直接调用即可。

public class MessageService : IMessageService, ITransientDependency
{
  private readonly IKafkaService _kafkaService;
  public MessageService(IKafkaService kafkaService)
  {
    _kafkaService = kafkaService;
  }

  public async Task RequestTraceAdded(XxxEventData eventData)
  {
    await _kafkaService.PublishAsync(eventData.TopicName, eventData);
  }
}

以上相当于一个生产者,当我们消息队列发出后,还需一个消费者进行消费,所以可以使用一个控制台项目接收消息来处理业务。

var cts = new CancellationTokenSource();
Console.CancelKeyPress += (_, e) =>
{
  e.Cancel = true;
  cts.Cancel();
};

await kafkaService.SubscribeAsync<XxxEventData>(topics, async (eventData) =>
{
  // Your logic

  Console.WriteLine($" - {eventData.EventTime:yyyy-MM-dd HH:mm:ss} 【{eventData.TopicName}】- > 已处理");
}, cts.Token);

IKafkaService 中已经写了订阅消息的接口,这里也是注入后直接使用即可。

生产者消费者示例

生产者

static async Task Main(string[] args)
{
  if (args.Length != 2)
  {
    Console.WriteLine("Usage: .. brokerList topicName");
    // 127.0.0.1:9092 helloTopic
    return;
  }

  var brokerList = args.First();
  var topicName = args.Last();

  var config = new ProducerConfig { BootstrapServers = brokerList };

  using var producer = new ProducerBuilder<string, string>(config).Build();

  Console.WriteLine("\n-----------------------------------------------------------------------");
  Console.WriteLine($"Producer {producer.Name} producing on topic {topicName}.");
  Console.WriteLine("-----------------------------------------------------------------------");
  Console.WriteLine("To create a kafka message with UTF-8 encoded key and value:");
  Console.WriteLine("> key value<Enter>");
  Console.WriteLine("To create a kafka message with a null key and UTF-8 encoded value:");
  Console.WriteLine("> value<enter>");
  Console.WriteLine("Ctrl-C to quit.\n");

  var cancelled = false;

  Console.CancelKeyPress += (_, e) =>
  {
    e.Cancel = true;
    cancelled = true;
  };

  while (!cancelled)
  {
    Console.Write("> ");

    var text = string.Empty;

    try
    {
      text = Console.ReadLine();
    }
    catch (IOException)
    {
      break;
    }

    if (string.IsNullOrWhiteSpace(text))
    {
      break;
    }

    var key = string.Empty;
    var val = text;

    var index = text.IndexOf(" ");
    if (index != -1)
    {
      key = text.Substring(0, index);
      val = text.Substring(index + 1);
    }

    try
    {
      var deliveryResult = await producer.ProduceAsync(topicName, new Message<string, string>
      {
        Key = key,
        Value = val
      });

      Console.WriteLine($"delivered to: {deliveryResult.TopicPartitionOffset}");
    }
    catch (ProduceException<string, string> e)
    {
      Console.WriteLine($"failed to deliver message: {e.Message} [{e.Error.Code}]");
    }
  }
}

消费者

static void Main(string[] args)
{
  if (args.Length != 2)
  {
    Console.WriteLine("Usage: .. brokerList topicName");
    // 127.0.0.1:9092 helloTopic
    return;
  }

  var brokerList = args.First();
  var topicName = args.Last();

  Console.WriteLine($"Started consumer, Ctrl-C to stop consuming");

  var cts = new CancellationTokenSource();
  Console.CancelKeyPress += (_, e) =>
  {
    e.Cancel = true;
    cts.Cancel();
  };

  var config = new ConsumerConfig
  {
    BootstrapServers = brokerList,
    GroupId = "consumer",
    EnableAutoCommit = false,
    StatisticsIntervalMs = 5000,
    SessionTimeoutMs = 6000,
    AutoOffsetReset = AutoOffsetReset.Earliest,
    EnablePartitionEof = true
  };

  const int commitPeriod = 5;

  using var consumer = new ConsumerBuilder<Ignore, string>(config)
             .SetErrorHandler((_, e) =>
             {
               Console.WriteLine($"Error: {e.Reason}");
             })
             .SetStatisticsHandler((_, json) =>
             {
               Console.WriteLine($" - {DateTime.Now:yyyy-MM-dd HH:mm:ss} > monitoring..");
               //Console.WriteLine($"Statistics: {json}");
             })
             .SetPartitionsAssignedHandler((c, partitions) =>
             {
               Console.WriteLine($"Assigned partitions: [{string.Join(", ", partitions)}]");
             })
             .SetPartitionsRevokedHandler((c, partitions) =>
             {
               Console.WriteLine($"Revoking assignment: [{string.Join(", ", partitions)}]");
             })
             .Build();
  consumer.Subscribe(topicName);

  try
  {
    while (true)
    {
      try
      {
        var consumeResult = consumer.Consume(cts.Token);

        if (consumeResult.IsPartitionEOF)
        {
          Console.WriteLine($"Reached end of topic {consumeResult.Topic}, partition {consumeResult.Partition}, offset {consumeResult.Offset}.");

          continue;
        }

        Console.WriteLine($"Received message at {consumeResult.TopicPartitionOffset}: {consumeResult.Message.Value}");

        if (consumeResult.Offset % commitPeriod == 0)
        {
          try
          {
            consumer.Commit(consumeResult);
          }
          catch (KafkaException e)
          {
            Console.WriteLine($"Commit error: {e.Error.Reason}");
          }
        }
      }
      catch (ConsumeException e)
      {
        Console.WriteLine($"Consume error: {e.Error.Reason}");
      }
    }
  }
  catch (OperationCanceledException)
  {
    Console.WriteLine("Closing consumer.");
    consumer.Close();
  }
}

到此这篇关于.NET Core下使用Kafka的方法步骤的文章就介绍到这了,更多相关.NET Core使用Kafka内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家!

您可能感兴趣的文章:
  • Kafka单节点伪分布式集群搭建实现过程详解
  • Docker搭建Zookeeper&Kafka集群的实现
  • Springboot 1.5.7整合Kafka-client代码示例
  • Spring boot 整合KAFKA消息队列的示例
  • Python confluent kafka客户端配置kerberos认证流程详解
  • docker-compose部署zk+kafka+storm集群的实现
  • Spring Boot集群管理工具KafkaAdminClient使用方法解析
  • Kafka多节点分布式集群搭建实现过程详解

相关文章

  • .NET Core下使用Kafka的方法步骤

    .NET Core下使用Kafka的方法步骤

    安装 CentOS安装 kafka Kafka : http://kafka.apache.org/downloads ZooLeeper : https://zookeeper.apache.org/releases.html 下载并解压 # 下载,并解压 $ wget
    2020-11-19
  • .NET Core下使用Log4Net记录日志的方法步骤

    .NET Core下使用Log4Net记录日志的方法步骤

    Log4Net 相信大家都很熟悉了,算是比较主流和著名的日志组件了。 官网: logging.apache.org 开源地址: https://github.com/apache/logging-log4net 最佳实践 在项
    2020-11-19
  • .NET CORE3.1实现微信小程序发送订阅消息

    .NET CORE3.1实现微信小程序发送订阅消息

    一、appsettings.json定义小程序配置信息 "WX": {   "AppId": "wx88822730803edd44",   "AppSecret": "75b269042e8b5026e6ed14aa24ba9353",   "Templates":
    2020-11-19
  • .Net Core3.0 配置Configuration的实现

    .Net Core3.0 配置Configuration的实现

    准备 .NET core和.NET项目配置上有了很大的改变,支持的也更加丰富了比如命令行,环境变量,内存中.NET对象,设置文件等等。.NET项目我们常常把配置信息放到webConf
    2020-11-19
  • 在.NET中使用DiagnosticSource的方法

    在.NET中使用DiagnosticSource的方法

    前言 DiagnosticSource是一个非常有意思的且非常有用的API,对于这些API它们允许不同的库发送命名事件,并且它们也允许应用程序订阅这些事件并处理它们,它使我们
    2020-11-19
  • .NET Core3.0创建Worker Services的实现

    .NET Core3.0创建Worker Services的实现

    .NET CORE 3.0新增了Worker Services的新项目模板,可以编写长时间运行的后台服务,并且能轻松的部署成windows服务或linux守护程序。如果安装的vs2019是中文版本,
    2020-11-19
  • 详解ASP.NET Core Web Api之JWT刷新Token

    详解ASP.NET Core Web Api之JWT刷新Token

    前言 如题,本节我们进入JWT最后一节内容,JWT本质上就是从身份认证服务器获取访问令牌,继而对于用户后续可访问受保护资源,但是关键问题是:访问令牌的生命周期到
    2020-11-19
  • ASP.NET Core 实现基本认证的示例代码

    ASP.NET Core 实现基本认证的示例代码

    HTTP基本认证 在HTTP中,HTTP基本认证(Basic Authentication)是一种允许网页浏览器或其他客户端程序以(用户名:口令) 请求资源的身份验证方式,不要求cookie,sess
    2020-11-19
  • .NET Core3.0 日志 logging的实现

    .NET Core3.0 日志 logging的实现

    多年的经验,日志记录是软件开发的重要组成部分。没有日志记录机制的系统不是完善的系统。在开发阶段可以通过debug附件进程进行交互调试,可以检测到一些问题,但是
    2020-11-19
  • ASP.NetCore使用Swagger实战

    ASP.NetCore使用Swagger实战

    踩坑不背锅,.NET Core 试深浅 关于Swagger什么是swagger所带来的帮助 使用Swagger 关于Swagger 什么是swagger 使人和计算机在看不到源码或者看不到文档或者不能通过
    2020-11-19

最新评论