Commit f8759398 authored by Savorboard's avatar Savorboard

fix multiple subscribing only received one message bug.

parent 2eaca664
using System; using System;
using System.Collections.Generic;
using System.Text; using System.Text;
using System.Threading; using System.Threading;
using Confluent.Kafka; using Confluent.Kafka;
...@@ -25,19 +26,18 @@ namespace DotNetCore.CAP.Kafka ...@@ -25,19 +26,18 @@ namespace DotNetCore.CAP.Kafka
StringDeserializer = new StringDeserializer(Encoding.UTF8); StringDeserializer = new StringDeserializer(Encoding.UTF8);
} }
public void Subscribe(string topic) public void Subscribe(IEnumerable<string> topics)
{ {
Subscribe(topic, 0); if (topics == null)
} throw new ArgumentNullException(nameof(topics));
public void Subscribe(string topicName, int partition)
{
if (_consumerClient == null) if (_consumerClient == null)
{ {
InitKafkaClient(); InitKafkaClient();
} }
_consumerClient.Assignment.Add(new TopicPartition(topicName, partition));
_consumerClient.Subscribe(topicName); //_consumerClient.Assign(topics.Select(x=> new TopicPartition(x, 0)));
_consumerClient.Subscribe(topics);
} }
public void Listening(TimeSpan timeout, CancellationToken cancellationToken) public void Listening(TimeSpan timeout, CancellationToken cancellationToken)
......
using System; using System;
using System.Collections.Generic;
using System.Threading; using System.Threading;
namespace DotNetCore.CAP namespace DotNetCore.CAP
...@@ -8,9 +9,7 @@ namespace DotNetCore.CAP ...@@ -8,9 +9,7 @@ namespace DotNetCore.CAP
/// </summary> /// </summary>
public interface IConsumerClient : IDisposable public interface IConsumerClient : IDisposable
{ {
void Subscribe(string topic); void Subscribe(IEnumerable<string> topics);
void Subscribe(string topic, int partition);
void Listening(TimeSpan timeout, CancellationToken cancellationToken); void Listening(TimeSpan timeout, CancellationToken cancellationToken);
......
using System; using System;
using System.Linq;
using System.Threading; using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using DotNetCore.CAP.Infrastructure; using DotNetCore.CAP.Infrastructure;
...@@ -56,10 +57,7 @@ namespace DotNetCore.CAP ...@@ -56,10 +57,7 @@ namespace DotNetCore.CAP
{ {
RegisterMessageProcessor(client); RegisterMessageProcessor(client);
foreach (var item in matchGroup.Value) client.Subscribe(matchGroup.Value.Select(x => x.Attribute.Name));
{
client.Subscribe(item.Attribute.Name);
}
client.Listening(_pollingDelay, _cts.Token); client.Listening(_pollingDelay, _cts.Token);
} }
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment