• 设为首页
  • 点击收藏
  • 手机版
    手机扫一扫访问
    迪恩网络手机版
  • 关注官方公众号
    微信扫一扫关注
    迪恩网络公众号

C# KafkaNet.Producer类代码示例

原作者: [db:作者] 来自: [db:来源] 收藏 邀请

本文整理汇总了C#中KafkaNet.Producer的典型用法代码示例。如果您正苦于以下问题:C# Producer类的具体用法?C# Producer怎么用?C# Producer使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。



Producer类属于KafkaNet命名空间,在下文中一共展示了Producer类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的C#代码示例。

示例1: Main

        public void Main(string[] args)
        {
            var options = GetOptions(args);
            if (options == null) return;

            var count = 0;
            var lastCount = 0;
            var reporter = new Task(() =>
            {
                while (true)
                {
                    var current = count;
                    Console.WriteLine("{0} messages in last second.", current - lastCount);
                    lastCount = current;
                    Thread.Sleep(1000);
                }
            });
            var kafkaOptions = new KafkaOptions(options.KafkaNodeUri);// { Log = new ConsoleLog() };
            using (var router = new BrokerRouter(kafkaOptions))
            using (var client = new KafkaNet.Producer(router))
            {
                reporter.Start();
                while (true)
                {
                    Thread.Sleep(100);
                    client.SendMessageAsync("TestHarness", new[] { new Message() { Value = BitConverter.GetBytes(DateTime.Now.Ticks) } });
                    count++;
                }
            }
        }
开发者ID:ducas,项目名称:KafkaProducer,代码行数:30,代码来源:Program.cs


示例2: Main

        static void Main(string[] args)
        {
            var options = new KafkaOptions(new Uri("http://CSDKAFKA01:9092"), new Uri("http://CSDKAFKA02:9092"))
                {
                    Log = new ConsoleLog()
                };
            var router = new BrokerRouter(options);
            var client = new Producer(router);

            Task.Factory.StartNew(() =>
                {
                    var consumer = new Consumer(new ConsumerOptions("TestHarness", router));
                    foreach (var data in consumer.Consume())
                    {
                        Console.WriteLine("Response: P{0},O{1} : {2}", data.Meta.PartitionId, data.Meta.Offset, data.Value);
                    }
                });

            Console.WriteLine("Type a message and press enter...");
            while (true)
            {
                var message = Console.ReadLine();
                if (message == "quit") break;
                client.SendMessageAsync("TestHarness", new[] {new Message {Value = message}});
            }

            using (client)
            using (router)
            {

            }
        }
开发者ID:ntent-ad,项目名称:kafka-net,代码行数:32,代码来源:Program.cs


示例3: EnsureProducerDisposesRouter

 public void EnsureProducerDisposesRouter()
 {
     var router = _kernel.GetMock<IBrokerRouter>();
     var producer = new Producer(router.Object);
     using (producer) { }
     router.Verify(x => x.Dispose(), Times.Once());
 }
开发者ID:ntent-ad,项目名称:kafka-net,代码行数:7,代码来源:ProducerTests.cs


示例4: EnsureGzipCanDecompressMessageFromKafka

        public void EnsureGzipCanDecompressMessageFromKafka()
        {
            var router = new BrokerRouter(_options);
            var producer = new Producer(router);

            var offsets = producer.GetTopicOffsetAsync(IntegrationConfig.IntegrationCompressionTopic).Result;

            var consumer = new Consumer(new ConsumerOptions(IntegrationConfig.IntegrationCompressionTopic, router) { PartitionWhitelist = new List<int>() { 0 } },
                offsets.Select(x => new OffsetPosition(x.PartitionId, x.Offsets.Max())).ToArray());
            int numberOfmessage = 3;
            for (int i = 0; i < numberOfmessage; i++)
            {
                producer.SendMessageAsync(IntegrationConfig.IntegrationCompressionTopic, new[] { new Message(i.ToString()) }, codec: MessageCodec.CodecGzip,
              partition: 0);
            }

            var results = consumer.Consume(new CancellationTokenSource(TimeSpan.FromMinutes(1)).Token).Take(numberOfmessage).ToList();

            for (int i = 0; i < numberOfmessage; i++)
            {
                Assert.That(results[i].Value.ToUtf8String(), Is.EqualTo(i.ToString()));
            }

            using (producer)
            using (consumer) { }
        }
开发者ID:gigya,项目名称:KafkaNetClient,代码行数:26,代码来源:GzipProducerConsumerTests.cs


示例5: Run

 public void Run()
 {
     var options = new KafkaOptions(new Uri("http://localhost:9092"));
     var router = new BrokerRouter(options);
     _producer = new KafkaNet.Producer(router);
     Process();
 }
开发者ID:gobixm,项目名称:learn,代码行数:7,代码来源:Producer.cs


示例6: ProducerShouldReportCorrectAmountOfAsyncRequests

        public void ProducerShouldReportCorrectAmountOfAsyncRequests()
        {
            var semaphore = new SemaphoreSlim(0);
            var routerProxy = new FakeBrokerRouter();
            //block the second call returning from send message async
            routerProxy.BrokerConn0.ProduceResponseFunction = () => { semaphore.Wait(); return new ProduceResponse(); };

            var router = routerProxy.Create();
            using (var producer = new Producer(router, maximumAsyncRequests: 1) { BatchSize = 1 })
            {
                var messages = new[] { new Message("1") };

                Assert.That(producer.AsyncCount, Is.EqualTo(0));

                var sendTask = producer.SendMessageAsync(BrokerRouterProxy.TestTopic, messages);

                TaskTest.WaitFor(() => producer.AsyncCount > 0);
                Assert.That(producer.AsyncCount, Is.EqualTo(1), "One async operation should be sending.");

                semaphore.Release();
                sendTask.Wait(TimeSpan.FromMilliseconds(500));

                Assert.That(sendTask.IsCompleted, Is.True, "Send task should be marked as completed.");
                Assert.That(producer.AsyncCount, Is.EqualTo(0), "Async should now show zero count.");
            }
        }
开发者ID:jsifantu,项目名称:kafka-net,代码行数:26,代码来源:ProducerTests.cs


示例7: ProtocolGateway

        public async Task ProtocolGateway()
        {
            int partitionId = 0;
            var router = new BrokerRouter(Options);

            var producer = new Producer(router);
            string messge1 = Guid.NewGuid().ToString();
            var respose = await producer.SendMessageAsync(IntegrationConfig.IntegrationTopic, new[] { new Message(messge1) }, 1, null, MessageCodec.CodecNone, partitionId);
            var offset = respose.FirstOrDefault().Offset;

            ProtocolGateway protocolGateway = new ProtocolGateway(IntegrationConfig.IntegrationUri);
            var fetch = new Fetch
                         {
                             Topic = IntegrationConfig.IntegrationTopic,
                             PartitionId = partitionId,
                             Offset = offset,
                             MaxBytes = 32000,
                         };

            var fetches = new List<Fetch> { fetch };

            var fetchRequest = new FetchRequest
                {
                    MaxWaitTime = 1000,
                    MinBytes = 10,
                    Fetches = fetches
                };

            var r = await protocolGateway.SendProtocolRequest(fetchRequest, IntegrationConfig.IntegrationTopic, partitionId);
            //  var r1 = await protocolGateway.SendProtocolRequest(fetchRequest, IntegrationConfig.IntegrationTopic, partitionId);
            Assert.IsTrue(r.Messages.FirstOrDefault().Value.ToUtf8String() == messge1);
        }
开发者ID:gabikliot,项目名称:KafkaNetClient,代码行数:32,代码来源:ProtocolGatewayTest.cs


示例8: FetchMessagesCacheContainsAllRequestTest

        public async Task FetchMessagesCacheContainsAllRequestTest()
        {
            // Creating a broker router and a protocol gateway for the producer and consumer
            var brokerRouter = new BrokerRouter(_options);
            var protocolGateway = new ProtocolGateway(_kafkaUri);

            Producer producer = new Producer(brokerRouter);
            ManualConsumer consumer = new ManualConsumer(_partitionId, _topic, protocolGateway, "TestClient", DefaultMaxMessageSetSize);

            var offset = await consumer.FetchLastOffset();

            // Creating 5 messages
            List<Message> messages = CreateTestMessages(10, 1);

            await producer.SendMessageAsync(_topic, messages, partition: _partitionId, timeout: TimeSpan.FromSeconds(3));

            // Now let's consume
            var result = (await consumer.FetchMessages(5, offset)).ToList();

            CheckMessages(messages.Take(5).ToList(), result);

            // Now let's consume again
            result = (await consumer.FetchMessages(5, offset + 5)).ToList();

            CheckMessages(messages.Skip(5).ToList(), result);
        }
开发者ID:pooyasoleimany,项目名称:KafkaNetClient,代码行数:26,代码来源:ManualConsumerTests.cs


示例9: ConsumerShouldConsumeInSameOrderAsProduced

        public void ConsumerShouldConsumeInSameOrderAsProduced()
        {
            var expected = new List<string> { "0", "1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12", "13", "14", "15", "16", "17", "18", "19" };
            var testId = Guid.NewGuid().ToString();

            using (var router = new BrokerRouter(new KafkaOptions(IntegrationConfig.IntegrationUri)))
            using (var producer = new Producer(router))
            {

                var offsets = producer.GetTopicOffsetAsync(IntegrationConfig.IntegrationTopic).Result;

                using (var consumer = new Consumer(new ConsumerOptions(IntegrationConfig.IntegrationTopic, router),
                    offsets.Select(x => new OffsetPosition(x.PartitionId, x.Offsets.Max())).ToArray()))
                {

                    for (int i = 0; i < 20; i++)
                    {
                        producer.SendMessageAsync(IntegrationConfig.IntegrationTopic, new[] { new Message(i.ToString(), testId) }).Wait();
                    }

                    var results = consumer.Consume().Take(20).ToList();

                    //ensure the produced messages arrived
                    Console.WriteLine("Message order:  {0}", string.Join(", ", results.Select(x => x.Value.ToUtf8String()).ToList()));

                    Assert.That(results.Count, Is.EqualTo(20));
                    Assert.That(results.Select(x => x.Value.ToUtf8String()).ToList(), Is.EqualTo(expected), "Expected the message list in the correct order.");
                    Assert.That(results.Any(x => x.Key.ToUtf8String() != testId), Is.False);
                }
            }
        }
开发者ID:jsifantu,项目名称:kafka-net,代码行数:31,代码来源:ProducerConsumerIntegrationTests.cs


示例10: ConsumerShouldBeAbleToSeekBackToEarlierOffset

        public void ConsumerShouldBeAbleToSeekBackToEarlierOffset()
        {
            var producer = new Producer(_router);

            var offsets = producer.GetTopicOffsetAsync("LoadTest").Result
                .Select(x => new OffsetPosition(x.PartitionId, x.Offsets.Max())).ToArray();

            var consumer = new Consumer(new ConsumerOptions("LoadTest", _router), offsets);

            var tasks = new List<Task<List<ProduceResponse>>>();
            for (int i = 0; i < 20; i++)
            {
                tasks.Add(producer.SendMessageAsync("LoadTest", new[] { new Message { Value = i.ToString(), Key = "1" } }));
            }
            Task.WaitAll(tasks.ToArray());

            var results = consumer.Consume().Take(20).ToList();

            //ensure the produced messages arrived
            for (int i = 0; i < 20; i++)
            {
                Assert.That(results[i].Value == i.ToString());
            }

            //seek back to initial offset
            consumer.SetOffsetPosition(offsets);

            //ensure all produced messages arrive again
            for (int i = 0; i < 20; i++)
            {
                Assert.That(results[i].Value == i.ToString());
            }
        }
开发者ID:jmucza,项目名称:kafka-net,代码行数:33,代码来源:ProducerConsumerTests.cs


示例11: ConsumerShouldBeAbleToGetCurrentOffsetInformation

        public void ConsumerShouldBeAbleToGetCurrentOffsetInformation()
        {
            var producer = new Producer(_router);

            var startOffsets = producer.GetTopicOffsetAsync("LoadTest").Result
                .Select(x => new OffsetPosition(x.PartitionId, x.Offsets.Max())).ToArray();

            var consumer = new Consumer(new ConsumerOptions("LoadTest", _router), startOffsets);

            var tasks = new List<Task<List<ProduceResponse>>>();
            for (int i = 0; i < 20; i++)
            {
                tasks.Add(producer.SendMessageAsync("LoadTest", new[] { new Message { Value = i.ToString(), Key = "1" } }));
            }
            Task.WaitAll(tasks.ToArray());

            var results = consumer.Consume().Take(20).ToList();

            //ensure the produced messages arrived
            for (int i = 0; i < 20; i++)
            {
                Assert.That(results[i].Value == i.ToString());
            }

            //the current offsets should be 20 positions higher than start
            var currentOffsets = consumer.GetOffsetPosition();
            Assert.That(currentOffsets.Sum(x => x.Offset) - startOffsets.Sum(x => x.Offset), Is.EqualTo(20));
        }
开发者ID:jmucza,项目名称:kafka-net,代码行数:28,代码来源:ProducerConsumerTests.cs


示例12: DestinationKafka

        public DestinationKafka(params Uri[] servers)
        {
            var options = new KafkaOptions(servers) { Log = new ConsoleLogger() };
            _router = new BrokerRouter(options);
            _producer = new Producer(_router, maximumMessageBuffer: 5000, maximumAsyncRequests: 10) { BatchSize = 1000, BatchDelayTime = TimeSpan.FromSeconds(1) };

            StatisticsTracker.OnStatisticsHeartbeat += StatisticsTracker_OnStatisticsHeartbeat;
        }
开发者ID:jsifantu,项目名称:kafka-net,代码行数:8,代码来源:DestinationKafka.cs


示例13: SendAsyncShouldGetOneResultForMessage

        public async void SendAsyncShouldGetOneResultForMessage()
        {
            using (var router = new BrokerRouter(new KafkaOptions(IntegrationConfig.IntegrationUri)))
            using (var producer = new Producer(router))
            {
                var result = await producer.SendMessageAsync(IntegrationConfig.IntegrationTopic, new[] { new Message(Guid.NewGuid().ToString()) });

                Assert.That(result.Count, Is.EqualTo(1));
            }
        }
开发者ID:BDeus,项目名称:KafkaNetClient,代码行数:10,代码来源:ProducerIntegrationTests.cs


示例14: SendAsyncShouldGetAResultForEachPartitionSentTo

        public async void SendAsyncShouldGetAResultForEachPartitionSentTo()
        {
            using (var router = new BrokerRouter(new KafkaOptions(IntegrationConfig.IntegrationUri)))
            using (var producer = new Producer(router))
            {
                var result = await producer.SendMessageAsync(IntegrationConfig.IntegrationTopic, new[] { new Message("1"), new Message("2"), new Message("3") });

                Assert.That(result.Count, Is.EqualTo(2));
            }
        }
开发者ID:jsifantu,项目名称:kafka-net,代码行数:10,代码来源:ProducerIntegrationTests.cs


示例15: ProducerAckLevel

 public async Task ProducerAckLevel()
 {
     using (var router = new BrokerRouter(new KafkaOptions(IntegrationConfig.IntegrationUri) { Log = IntegrationConfig.NoDebugLog }))
     using (var producer = new Producer(router))
     {
         var responseAckLevel0 = await producer.SendMessageAsync(new Message("Ack Level 0"), IntegrationConfig.IntegrationTopic, acks: 0, partition: 0);
         Assert.AreEqual(responseAckLevel0.Offset, -1);
         var responseAckLevel1 = await producer.SendMessageAsync(new Message("Ack Level 1"), IntegrationConfig.IntegrationTopic, acks: 1, partition: 0);
         Assert.That(responseAckLevel1.Offset, Is.GreaterThan(-1));
     }
 }
开发者ID:pooyasoleimany,项目名称:KafkaNetClient,代码行数:11,代码来源:ProducerConsumerIntegrationTests.cs


示例16: GetProducer

 public static Producer GetProducer(this Kafka kafkaObj)
 {
     if (_producer == null)
     {
         var addresses = from x in kafkaObj.brokers
                         select new Uri(x.address);
         var router = new BrokerRouter(new KafkaOptions(addresses.ToArray()));
         _producer = new Producer(router);
     }
     return _producer;
 }
开发者ID:KunalSaini,项目名称:NLog.Kafka,代码行数:11,代码来源:KafkaClient.cs


示例17: ProducerAckLevel1ResponseOffsetShouldBeEqualToLastOffset

 public async Task ProducerAckLevel1ResponseOffsetShouldBeEqualToLastOffset()
 {
     using (var router = new BrokerRouter(new KafkaOptions(IntegrationConfig.IntegrationUri) { Log = IntegrationConfig.NoDebugLog }))
     using (var producer = new Producer(router))
     {
         var responseAckLevel1 = await producer.SendMessageAsync(new Message("Ack Level 1"), IntegrationConfig.IntegrationTopic, acks: 1, partition: 0);
         var offsetResponse = await producer.GetTopicOffsetAsync(IntegrationConfig.IntegrationTopic);
         var maxOffset = offsetResponse.Find(x => x.PartitionId == 0);
         Assert.AreEqual(responseAckLevel1.Offset, maxOffset.Offsets.Max() - 1);
     }
 }
开发者ID:pooyasoleimany,项目名称:KafkaNetClient,代码行数:11,代码来源:ProducerConsumerIntegrationTests.cs


示例18: ManualConsumerFailure

        public async Task ManualConsumerFailure()
        {
            string topic = "TestTopicIssue13-3R-1P";
            var manualConsumer = new ManualConsumer(0, topic, new ProtocolGateway(_options), "test client", 10000);
            long offset = await manualConsumer.FetchLastOffset();

            var router = new BrokerRouter(_options);
            var producer = new Producer(router);
            SandMessageForever(producer, topic);
            await ReadMessageForever(manualConsumer, offset);
        }
开发者ID:gigya,项目名称:KafkaNetClient,代码行数:11,代码来源:ManualTesting.cs


示例19: Main

        static void Main(string[] args)
        {
            const string topicName = "kafka.SimpleTopic";
            TopicClient topicClient = new TopicClient();
            TopicView topicView = topicClient.getTopic(topicName);
            List<Uri> brokers = getBrokerList(topicView);
            //create an options file that sets up driver preferences
            var options = new KafkaOptions()
            {
                Log = new ConsoleLog()
            };
            options.KafkaServerUri = brokers;

            //start an out of process thread that runs a consumer that will write all received messages to the console
            Task.Run(() =>
            {
                var consumer = new Consumer(new ConsumerOptions(topicName, new BrokerRouter(options)) { Log = new ConsoleLog() });
                foreach (var data in consumer.Consume())
                {
                    Console.WriteLine("Response: P{0},O{1} : {2}", data.Meta.PartitionId, data.Meta.Offset, data.Value.ToUtf8String());
                }
            });

            //create a producer to send messages with
            var producer = new KafkaNet.Producer(new BrokerRouter(options))
            {
                BatchSize = 100,
                BatchDelayTime = TimeSpan.FromMilliseconds(2000)
            };


            //take in console read messages
            Console.WriteLine("Type a message and press enter...");
            while (true)
            {
                var message = Console.ReadLine();
                if (message == "quit") break;

                if (string.IsNullOrEmpty(message))
                {
                    //send a random batch of messages
                    SendRandomBatch(producer, topicName, 200);
                }
                else
                {
                    producer.SendMessageAsync(topicName, new[] { new Message(message) });
                }
            }

            using (producer)
            {

            }
        }
开发者ID:qmwu2000,项目名称:hermes,代码行数:54,代码来源:Program.cs


示例20: Send

        public async Task Send(object channel, Address address, TransportMessage message, object properties)
        {
            var options = new KafkaOptions(new Uri(address.Machine));
            var router = new BrokerRouter(options);
            var topic = address.Queue;
            var messageString = System.Text.Encoding.Default.GetString(message.Body);

            using (var client = new Producer(router))
            {
                await client.SendMessageAsync(topic, new[] { new Message(messageString) });
            }
        }
开发者ID:colin-higgins,项目名称:NServiceBus.Kafka,代码行数:12,代码来源:SimpleTopology.cs



注:本文中的KafkaNet.Producer类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。


鲜花

握手

雷人

路过

鸡蛋
该文章已有0人参与评论

请发表评论

全部评论

专题导读
上一篇:
C# Type.KecaknoahObject类代码示例发布时间:2022-05-26
下一篇:
C# KacTalk.ktString类代码示例发布时间:2022-05-26
热门推荐
阅读排行榜

扫描微信二维码

查看手机版网站

随时了解更新最新资讯

139-2527-9053

在线客服(服务时间 9:00~18:00)

在线QQ客服
地址:深圳市南山区西丽大学城创智工业园
电邮:jeky_zhao#qq.com
移动电话:139-2527-9053

Powered by 互联科技 X3.4© 2001-2213 极客世界.|Sitemap