微软云消息队列 Azure service bus queue

shy-huang 2018-05-07 原文

前言

第一次使用消息队列，遇到了一些问题：同一个消息有多次出列。是一个消息只入列一次，还是多次？还是因为出列问题，出列了多次？

Microsoft Azure service bus queue

Azure service bus queue在Azure上创建一个service bus，在service bus 上创建一个 queue，创建的时候注意 enable duplicate detected message这个选项选上，防止消息重复入列。

找到SAS Policy: RootManageSharedAccessKey 的值，复制下来，用作connectstring。

Azure service bus sample on github

配置servicebus 的queue

namespace：Microsoft.Azure.ServiceBus;

初始化queueClient:

private static readonly Serilog.ILogger loggerCore = LoggerCoreFactory.GetLoggerCore();
const string ServiceBusConnectionString = "Endpoint=sb://{YOUR-NAMESPACE-ON-Azure}.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey={YOUR-KEYS}";
const string QueueName = "{YOUR-QUEUE-NAME}";
static IQueueClient queueClient;

static MessageQueueHandler()
{
    queueClient = new QueueClient(ServiceBusConnectionString, QueueName);
}

消息入列

这里添加messageID 作为入列依据，防止同一消息多次入列，然后不用频繁的关闭消息连接，影响性能。

public static async Task MessageEnqueueAsync(string message)
{
    // Send messages.
    await SendMessagesAsync(message);
    // don't close frequently 
    //await queueClient.CloseAsync();
}

static async Task SendMessagesAsync(string messageXML)
{
    try
    {
        // Create a new message to send to the queue.
        var messageStr = ParseMessageType.Parse(messageXML);
        var msgId = messageStr.Body.MsgId.Value;
        var message = new Microsoft.Azure.ServiceBus.Message
        {
            MessageId = msgId,// avoid same message enqueue more than once
            Body = Encoding.UTF8.GetBytes(messageXML),
        };
        loggerCore.Information($"message enqueue:{messageXML}");
        // Send the message to the queue.
        await queueClient.SendAsync(message);
    }
    catch (Exception exception)
    {
        loggerCore.Error($"message enqueue error:{exception.ToString()}");
    }
}

出列

public static async Task MessageDequeueAsync()
{
    // please choice PeekLock mode
    queueClient = new QueueClient(ServiceBusConnectionString, QueueName, ReceiveMode.PeekLock);

    // Register the queue message handler and receive messages in a loop
    queueClient.RegisterMessageHandler(
    async (message, token) =>
    {
        // Process the message
        await PostMessageToBot(message);
        loggerCore.Information($"Received message: SequenceNumber:{message.SystemProperties.SequenceNumber} Body:{Encoding.UTF8.GetString(message.Body)}");
        // Complete the message so that it is not received again.
        // This can be done only if the queueClient is opened in ReceiveMode.PeekLock mode.
        await queueClient.CompleteAsync(message.SystemProperties.LockToken);
    },
    async (exceptionEvent) =>
    {
        // Process the exception
        loggerCore.Error($"WeChat message dequeue exception:{exceptionEvent.ToString()}");
    });
}

这样消息就会根据入列的先后，逐次出列。