如何将消息从一个节点发送到另一个节点

本文关键字：节点 一个 另一个 消息 | 更新日期: 2023-09-27 18:13:22

我想发送eventhub客户，然后将其下载到天气等样本数据并发送另一个eventhub。我的代码不能正常工作。没有错误，但是数据没有发送到数据库。

public Task ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> messages)
        {
                try
            {
                foreach (EventData message in messages)
                {
                        string data = Encoding.UTF8.GetString(message.GetBytes());
                        NewClient Client = JsonConvert.DeserializeObject<NewClient>(data);
                    if (Client.City != null && Client.Street != null )
                    {
                        GoogleGeoApi GeoClient = new GoogleGeoApi();
                        GeoClient.SetCoordinates(Client.City, Client.Street);
                        WeatherApi WeatherApiobject = new WeatherApi();
                        WeatherApiobject.GetJson(GeoClient.convertlat, GeoClient.convertlng);
                        string weatherdata = WeatherApiobject.sendEvent;
                        SenderEvent NewSenderEvent = new SenderEvent();
                        NewSenderEvent.DataSender(weatherdata, ConstFile.WeatherEventHubName);
                        //StartH(ConstFile.WeatherEventHubName).Wait();
                    }
                    Interlocked.Increment(ref this.totalMessages);
                    this.LastMessageOffset = message.Offset;
                }
                if (this.IsClosed)
                {
                    this.IsReceivedMessageAfterClose = true;
                }
                if (this.checkpointStopWatch.Elapsed > TimeSpan.FromMinutes(1))
                {
                    lock (this)
                    {
                        this.checkpointStopWatch.Reset();
                        return context.CheckpointAsync();
                    }
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine("{0} > Event Hub Exception: {1}", DateTime.Now, ex.Message);
            }
            return Task.FromResult<object>(null);
        }

我将添加我的接收器甚至看起来像这样:https://code.msdn.microsoft.com/Service-Bus-Event-Hub-45f43fc3/view/SourceCode内容

如何将消息从一个节点发送到另一个节点

你可以使用流分析很容易做到这一点!你的第一个事件中心是流分析的输入。然后你可以对流写一个查询(Select * from [Input1]…这会让你得到一切)。您可以将流输出回另一个Event Hub。

 public class SenderEvent
    {
        public void DataSender(string data, string eventhubname)
        {
            var eventhubclient = EventHubClient.CreateFromConnectionString(ConstFile.eventHubConnectionString, eventhubname);       
            EventData data1 = new EventData(Encoding.UTF8.GetBytes(data));
            eventhubclient.Send(data1);
        }

private static async Task StartHost(string eventHubName)
        {
            string eventProcessorHostName = "1";
            string storageConnectionString = string.Format("DefaultEndpointsProtocol=https;AccountName={0};AccountKey={1}", ConstFile.storageAccountName, ConstFile.storageAccountKey);
            host = new EventProcessorHost(
                eventProcessorHostName,
                eventHubName,
                ConstFile.ConsumerGroup,
                ConstFile.eventHubConnectionString,
                storageConnectionString, eventHubName.ToLowerInvariant());
            factory = new DemoEventProcessorFactory(eventProcessorHostName);
            try
            {
                var options = new EventProcessorOptions();
                options.ExceptionReceived += (sender, e) => { Console.WriteLine(e.Exception); };
                await host.RegisterEventProcessorFactoryAsync(factory);
            }
            catch (Exception exception)
            {
                Console.ForegroundColor = ConsoleColor.Red;
                Console.WriteLine("{0} > Exception: {1}", DateTime.Now.ToString(), exception.Message);
                Console.ResetColor();
            }
        }
    }
}

这取决于:

  1. 如果你的事件中心是低延迟-你可以使用事件中心触发器(azure函数，该函数将在事件中心接收到新的事件)。在你的azure函数中，你使用的代码如示例

  2. 如果你有很多事件-你可以尝试azure流分析系统过滤出你的天气数据，并将其发送到不同的活动中心。您可以像这样构建一个管道:

    EventHub1 -> AzureStreamAnalytics -> EventHub2

相关文章：