CSharpClient-for-Kafka

Apache Kafka协议的.Net实现,它通过Producer/Consumer类提供基本功能。 该项目还提供平衡的消费者实施。(.Net implementation of the Apache Kafka Protocol that provides basic functionality through Producer/Consumer classes. The project also offers balanced consumer implementation.)

Github stars Tracking Chart

This project will be archived to be in read-only mode as no new development/maintenance is expected. Please look at the Note section below for alternate libraries.

Note

This library was created for Kafka 0.8 with an intention to have a native library built from scratch.
With Kafka protocol getting updated frequently with new features (which is expected until it reaches version 1.0), it doesn't seem beneficial to maintain a library built from scratch.
The right approach (and as suggested by Confluent) for now would be to use a C# wrapper around the librdkafka C-Library, which the confluent-kafka-dotnet client is doing.

So, if you are using Kafka 0.9 or higher, please move to using the confluent-kafka-dotnet client library.

CSharpClient-for-Kafka

Join the chat at https://gitter.im/Microsoft/Kafkanet
.Net implementation of the Apache Kafka Protocol that provides basic functionality through Producer/Consumer classes. The project also offers balanced consumer implementation.
The project is a fork from ExactTarget's Kafka-net Client.

Build CSharpClient-for-Kafka

  • Clone CSharpClient-for-Kafka through git clone https://github.com/Microsoft/CSharpClient-for-Kafka.git
  • Open src\KafkaNETLibraryAndConsole.sln in Visual Studio
  • Build Solution

Run Unit Tests

  • Open Test Window in Visual Studio: Test>Windows>Test Explorer
  • Run all

Using Console

  • Setup local Kafka and Zookeeper

Console Options

	topic                           Dump topics metadata, such as: earliest/latest offset, replica, ISR.
    consumesimple                   Consume data in single thread.
    consumegroup                    Monitor latest offset gap and speed of consumer group.
    consumegroupmonitor             Monitor latest offset gap and speed of consumer group.
    producesimple                   Produce data in single thread.
    produceperftest                 Produce data in multiple thread.
    eventserverperftest             Http Post data to event server in multiple thread.
    producemonitor                  Monitor latest offset.
    test                            Run some adhoc test cases.

Using the library

Producer

The Producer can send one message or an entire batch to Kafka. When sending a batch you can send to multiple topics at once

Producer Usage

var brokerConfig = new BrokerConfiguration()
{
    BrokerId = this.brokerId,
    Host = this.kafkaServerName,
    Port = this.kafkaPort
};
var config = new ProducerConfiguration(new List<BrokerConfiguration> { brokerConfig });
kafkaProducer = new Producer(config);
// here you construct your batch or a single message object
var batch=ConstructBatch();
kafkaProducer.Send(batch);

Simple Consumer

The simple Consumer allows full control for retrieving data. You could instantiate a Consumer directly by providing a ConsumerConfiguration and then calling Fetch.
CSharpClient-for-Kafka has a higher level wrapper around Consumer which allows consumer reuse and other benefits

Consumer Usage

// create the Consumer higher level manager
var managerConfig = new KafkaSimpleManagerConfiguration()
{
    FetchSize = FetchSize,
    BufferSize = BufferSize,
    Zookeeper = m_zookeeper
};
m_consumerManager = new KafkaSimpleManager<int, Kafka.Client.Messages.Message>(managerConfig);
// get all available partitions for a topic through the manager
var allPartitions = m_consumerManager.GetTopicPartitionsFromZK(m_topic);
// Refresh metadata and grab a consumer for desired partitions
m_consumerManager.RefreshMetadata(0, m_consumerId, 0, m_topic, true);
var partitionConsumer = m_consumerManager.GetConsumer(m_topic, partitionId);

Balanced Consumer

The balanced consumer manages partition assignment for each instance in the same consumer group. Rebalance are triggered by zookeeper changes.

Balanced Consumer Usage

// Here we create a balanced consumer on one consumer machine for consumerGroupId. All machines consuming for this group will get balanced together
ConsumerConfiguration config = new ConsumerConfiguration
{
    AutoCommit = false,
    GroupId = consumerGroupId
    ConsumerId = uniqueConsumerId
    MaxFetchBufferLength = m_BufferMaxNoOfMessages,
    FetchSize = fetchSize,
    AutoOffsetReset = OffsetRequest.LargestTime,
    NumberOfTries = 20,
    ZooKeeper = new ZooKeeperConfiguration(zookeeperString, 30000, 30000, 2000)
};
var balancedConsumer = new ZookeeperConsumerConnector(config, true, m_ConsumerRebalanceHandler, m_ZKDisconnectHandler, m_ZKExpireHandler);
// grab streams for desired topics 
var streams = m_ZooKeeperConsumerConnector.CreateMessageStreams(m_TopicMap, new DefaultDecoder());
var KafkaMessageStream = streams[m_Topic][0];
// start consuming stream
foreach (Message message in m_KafkaMessageStream.GetCancellable(cancellationTokenSource.Token))
....

Contribute

Contributions to CSharpClient-for-Kafka are welcome. Here is how you can contribute to CSharpClient-for-Kafka:

Main metrics

Overview
Name With Ownermicrosoft/CSharpClient-for-Kafka
Primary LanguageC#
Program languageC# (Language Count: 1)
PlatformWindows
License:Other
所有者活动
Created At2016-02-18 22:41:46
Pushed At2018-10-11 21:14:08
Last Commit At2018-10-11 14:14:08
Release Count0
用户参与
Stargazers Count463
Watchers Count99
Fork Count95
Commits Count96
Has Issues Enabled
Issues Count45
Issue Open Count33
Pull Requests Count31
Pull Requests Open Count10
Pull Requests Close Count5
项目设置
Has Wiki Enabled
Is Archived
Is Fork
Is Locked
Is Mirror
Is Private