EasyNetQ简单使用手册

github

1
https://github.com/EasyNetQ/EasyNetQ

docker

1
2
3
4
5
docker pull rabbitmq:management
docker run -d --restart=always --hostname rabbit --name rabbit -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -p 15672:15672 -p 5672:5672 -p 25672:25672 -p 61613:61613 -p 1883:1883 rabbitmq:management
docker run -d --restart=always --hostname rabbit --name netcore-rabbit-200 --net zjx-network --ip 172.18.0.200 -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin rabbitmq:management
## 15672：控制台端口号
## 5672：应用访问端口号

nuget

1
dotnet add package EasyNetQ --version 6.3.1

netcore

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
services.AddSingleton(RabbitHutch.CreateBus("host=localhost;username=admin;password=admin;timeout=60"));
app.UseSubscribe("OrderService", Assembly.GetExecutingAssembly());

public static class EApplicationExtenssion
{
    public static IApplicationBuilder UseSubscribe(this IApplicationBuilder appBuilder, string subscriptionIdPrefix, Assembly assembly)
    {
        var services = appBuilder.ApplicationServices.CreateScope().ServiceProvider;
        var lifeTime = services.GetService<Microsoft.Extensions.Hosting.IHostApplicationLifetime>();
        var bus = services.GetService<IBus>();
        lifeTime.ApplicationStarted.Register(() =>
        {
            var subscriber = new AutoSubscriber(bus, subscriptionIdPrefix);
            var ass = new Assembly[] { assembly };
            subscriber.Subscribe(ass);
            subscriber.SubscribeAsync(ass);
        });
        lifeTime.ApplicationStopped.Register(() => { bus.Dispose(); });
        return appBuilder;
    }
}

public class TestController : DefaultController
{
    private IBus bus;
    public TestController(EFDataContext ctx, IBus bus)
    {
        this.ctx = ctx;
        this.bus = bus;
    }
    public ActionResult Index()
    {
        // bus.PubSub.Publish(new Order
        // {
        //     OrderId = 999
        // });
        bus.Scheduler.FuturePublish(new Order
        {
            OrderId = 999
        }, TimeSpan.FromSeconds(5));
        return Success();
    }
}
[Queue("Qka.Order", ExchangeName = "Qka.Order")]
public class Order
{
    public int OrderId { get; set; }
}
public class OrderConsumer : IConsume<Order>
{
    [AutoSubscriberConsumer(SubscriptionId = "OrderService")]
    public void Consume(Order message, CancellationToken cancellationToken = default)
    {
        //业务代码
    }
}