渭南网站建设哪家好,电影网站空间配置,小程序怎么开发自己的小程序游戏,网络商城是什么意思最近维护的一个系统并发有点高#xff0c;所以想引入一个消息队列来进行削峰。考察了一些产品#xff0c;最终决定使用kafka来当做消息队列。以下是关于kafka的一些知识的整理笔记。kafkakafka 是分布式流式平台。它由linkedin开发#xff0c;后贡献给了Apache开源组织并成为… 最近维护的一个系统并发有点高所以想引入一个消息队列来进行削峰。考察了一些产品最终决定使用kafka来当做消息队列。以下是关于kafka的一些知识的整理笔记。kafkakafka 是分布式流式平台。它由linkedin开发后贡献给了Apache开源组织并成为顶级开源项目。它可以应用在高并发场景下的日志系统也可以当作消息队列来使用也可以当作消息服务对系统进行解耦。流处理平台有以下三种特性:1.可以让你发布和订阅流式的记录。这一方面与消息队列或者企业消息系统类似。2.可以储存流式的记录并且有较好的容错性。3.可以在流式记录产生时就进行处理。一般它可以应用于两个场景1.构造实时流数据管道它可以在系统或应用之间可靠地获取数据。(相当于message queue)2.构建实时流式应用程序对这些流数据进行转换或者影响。(就是流处理通过kafka stream topic和topic之间内部进行变化)brokerkafka中的每个节点即每个服务器就是一个broker 。topickafka中的topic是一个分类的概念表示一类消息。生产者在生产消息的时候需要指定topic消费者在消费消息的时候也需要指定topic。partitionpartition是分区的概念。kafka的一个topic可以有多个partition。每个partition会分散到不同的broker上起到负载均衡的作用。生产者的消息会通过算法均匀的分散在各个partition上。consumer groupkafka的消费者有个组的概念。一个partition可以被多consumer group订阅。每个消息会广播到每一个group中。但是每个消息只会被group中的一个consumer消费。相当于每个group一个partition只能有一个consumer订阅所以group中的consumer数量不可以超过topic中partition的数量。并且消息的消费的顺序在每个partition中是保证有序的但是在多个partition之间是不保证的因为consumer的消费速度是有快慢的。所以如果要用kafka实现严格的消息队列点对点模式那么我们可以设置一个partition并且设置一个consumer。如果对消息消费的顺序不是那么敏感那么可以设置多个partition来并行消费消息提高吞吐量。安装kafka为了能体验下kafka我们还是要实际安装一下kafka毕竟空想是没有用的。现在有了docker安装起来也是相当滴简单。我们只需要定义好docker-compose的yml就行了。version: 3
services:zookeeper:image: wurstmeister/zookeeperports:- 2181:2181kafka:image: wurstmeister/kafkadepends_on:- zookeeperports:- 9092:9092environment:KAFKA_ADVERTISED_HOST_NAME: 192.168.0.117KAFKA_CREATE_TOPICS: test:3:1KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181我们在yml里定义2个service1.zookeeperkafka的分布式依赖zookeeper所以我需要先定义它。2.kafka kafka的定义有几个地方要注意的。•depends_on:zookeeper 指定kafka依赖zookeeper这个service当启动kafka的时候自动会启动zookeeper。•KAFKA_ADVERTISED_HOST_NAME 这里要指定宿主机的ip•KAFKA_CREATE_TOPICS 这个变量只是的默认创建的topic。test:3:1代表创建一个名为test的topic并且创建3个分区1个复制。定义好这些之后我们只需要使用docker-compose命令运行它sudo docker-compose up -d
.net 操作 kafka安装好kafka的docker环境之后下面演示下如何使用.net操作kafka进行消息的生产与消费。生产者 static async Task Main(string[] args){Console.WriteLine(Hello World Producer!);var config new ProducerConfig{BootstrapServers 192.168.0.117:9092,ClientId Dns.GetHostName(),};using (var producer new ProducerBuilderNull, string(config).Build()){string topic test;for (int i 0; i 100; i){var msg message i;Console.WriteLine($Send message: value {msg});var result await producer.ProduceAsync(topic, new MessageNull, string { Value msg });Console.WriteLine($Result: key {result.Key} value {result.Value} partition:{result.TopicPartition});Thread.Sleep(500);}}Console.ReadLine();}
新建一个控制台项目从nuget安装kafka的官方client。Install-Package Confluent.Kafka
代码非常简单使用ProducerBuilder构造一个producer然后调用ProduceAsync方法发送消息。其中需要注意的是如果你的场景并发非常之高官方文档推荐的方法是Produce而不是ProduceAsync。这是一个比较迷的地方。按常理使用ProduceAsync应该比使用同步方法Produce能获得更高的并发才对。但是文档确确实实说高并发场景请使用Produce。可能是为了避免ProduceAsync结果返回的时候异步线程上下文切换造成的性能开销。原文There are a couple of additional benefits of using the Produce method. First, notification of message delivery (or failure) is strictly in the order of broker acknowledgement. With ProduceAsync, this is not the case because Tasks may complete on any thread pool thread. Second, Produce is more performant because there is unavoidable overhead in the higher level Task based API.
消费者 static void Main(string[] args){Console.WriteLine(Hello World kafka consumer !);var config new ConsumerConfig{BootstrapServers 192.168.0.117:9092,GroupId foo,AutoOffsetReset AutoOffsetReset.Earliest};var cancel false;using (var consumer new ConsumerBuilderIgnore, string(config).Build()){var topic test;consumer.Subscribe(topic);while (!cancel){var consumeResult consumer.Consume(CancellationToken.None);Console.WriteLine($Consumer message: { consumeResult.Message.Value} topic: {consumeResult.Topic} Partition: {consumeResult.Partition});}consumer.Close();}}
消费者的演示代码同样很简单。我们需要指定groupId然后订阅topic。使用ConsumerBuilder构造一个consumer然后调用Consume方法进行消费就可以。注意这里默认是自动commit消费。你也可以根据情况手动提交commit。运行一下我们运行一个生产者进程按照500ms的速度生产消息。运行三个consumer进行消费可以看到消息被均匀的推送到三个consumer上去。总结以上简单的介绍了kafka的背景、安装方法、使用场景。还简单演示了如何使用.net来操作kafka。它可以当作流式计算平台来使用也可以当作传统的消息队列使用。它当前非常流行网上的资料也多如牛毛。官方也提供了简单易用的.net sdk 为.net 平台集成kafka提供了便利。关注我的公众号一起玩转技术
