Larva.Messaging 1.0.0

消息类库

基于消息队列的消息库

• 支持消息路由、顺序消费；
• 支持部署多台消费端程序（相同Exchange下，一个队列仅允许被一个消费端程序消费；但一个消费端程序可以同时消费多个队列）；
• 采用Envelope<T>进行消息封装，通过实现接口IMessageHandler<T>进行消息处理；
• 消息发送，默认发送到mq失败时会不断重试，也可以通过初始化TopicSender时显式指定重试次数；
• 同一个队列，如果消息处理失败，则会不断重试，直到显式标记为丢弃消息(通过IMessageTransportationContext.DropMessage)；
• 支持专题模式（Topic）和发布订阅模式（Pubsub）；
• 发送端和消息端使用的消息类之间解耦，可以通过MessageTypeAttribute对消息类进行消息类型声明，默认为带名字空间的类名；
• 消费端反序列化失败，或者找不到对应的消息处理器时，利用rabbitmq的DLX特性，丢到一个DLX Exchange（使用默认Exchange）中，DLX Exchange发送的队列的命名=Exchange名 + "-dlx"，发布订阅模式需要手工建；

调用示例

• 定义消息

// 可以定义消息类型名，不设置，默认取带名字空间的类名
//[MessageType("Tests.Messages.1")]
public class Message01
{
    public Guid HostingFilialeId { get; set; }

    public Message01() { }

    public Message01(Guid hostingFilialeId)
    {
        HostingFilialeId = hostingFilialeId;
    }
}

• 消息发送

// 创建 Rabbitmq 连接（此为全局变量，一个应用共享一个连接即可）
var conn = new Connection(new ConnectionConfig(new Uri("amqp://user:pwd@host:port/virtual_host")));

// 创建Topic发送者（此为全局变量，一个对象实例对应使用一个rabbitmq的exchange）
ITopicSender publisher = new TopicSender(conn, exchangeName, queueCount);
//如果采用发布订阅方式，则如下声明
//IPubsubSender publisher = new PubsubSender(conn, exchangeName);

// 发送消息
var message = new Message01(new Guid("58437EDC-87B7-4995-A5C0-BB5FD0FE49E0"));
publisher.SendMessage(Envelope.Create(message, message.HostingFilialeId.ToString()));

// 释放
publisher.Dispose();

• 消息消费

// 定义消息处理器
// 注意：一个处理器类，可以定义为多个消息类型的处理器；但针对同一个Exchange，一个消息类型仅允许对应一个消息处理器
public class MessageHandler01 : IMessageHandler<Message01>, IMessageHandler<Message02>
{
    public void Handle(Message01 message, IMessageTransportationContext context)
    {
        if (context.RetryCount > 3)
        {
            // 执行次数超过3次，则记录日志
            context.DropMessage();// 丢弃消息
            return;
        }
        // 处理消息 Message01
    }

    public void Handle(Message02 message, IMessageTransportationContext context)
    {
        if (context.RetryCount > 3)
        {
            // 执行次数超过3次，则记录日志
            context.DropMessage();// 丢弃消息
            return;
        }
        // 处理消息 Message02
    }
}

// 创建 Rabbitmq 连接（此为全局变量，一个应用共享一个连接即可）
var conn = new Connection(new ConnectionConfig(new Uri("amqp://user:pwd@host:port/virtual_host")));

// 创建接受者
ITopicReceiver receiver = new TopicReceiver(conn, exchangeName, queueCount);
//如果采用发布订阅方式，则如下声明
//IPubsubReceiver receiver = new PubsubReceiver(conn, queueName);

// 注册消息处理器
receiver.RegisterMessageHandler<MessageHandler01>();
//如果按指定程序集下注册所有的消息处理器，可以如下调用
//receiver.RegisterMessageHandlerByAssembly(typeof(MessageHandler01).Assembly);

// 订阅所有队列
receiver.SubscribeAll();
// 如果采用分布式部署多台消费端后端程序，则通过指定队列序号来实现
// 注意：同一个队列，仅允许一个后端程序；但一个后端程序可以消费多个队列
// receiver.Subscribe(0);
//如果采用发布订阅方式，则如下调用
//receiver.Subscribe();

更新历史