接上一篇：ActiveMQ基础教程（三）：C#连接使用ActiveMQ消息队列
这里继续说下.net core集成使用ActiveMQ。因为代码比较多，所以放到gitee上：https://gitee.com/shanfeng1000/dotnetcore-demo/tree/master/Activemq
感兴趣的可以克隆下来再按照自己的需求修改，这里简单介绍一下使用的Demo（Demo基于.net core3.1的版本，其他版本可能需要自行测试）
生产者（AspNetCore.WebApi.Producer）
在Startup中添加相关服务项：
public void ConfigureServices(IServiceCollection services)
{
var brokerUris = new string[] { "192.168.209.133:61616", "192.168.209.134:61616", "192.168.209.135:61616" };
string userName = "test";
string password = "123456"; #region 日志记录 services.AddLogging(builder =>
{
builder.SetMinimumLevel(LogLevel.Trace);
});
services.AddActiveLogger(options =>
{
options.IsCluster = true;
options.ApplicationName = "WebApi";
options.BrokerUris = brokerUris;
options.Category = "Home";
options.UseQueue = false;
options.Destination = "logger";
options.MinLevel = LogLevel.Warning;
options.InitializeCount = 10;
options.IsPersistent = true;
options.Password = password;
options.UserName = userName;
}); #endregion
#region Active services.AddActiveProducer("active.queue", options =>
{
options.IsCluster = true;
options.BrokerUris = brokerUris;
options.Destination = "active.queue";
options.IsPersistent = true;
options.Transactional = false;
options.Password = password;
options.UserName = userName;
});
services.AddActiveProducer("active.topic", options =>
{
options.IsCluster = true;
options.BrokerUris = brokerUris;
options.Destination = "active.topic";
options.IsPersistent = true;
options.Transactional = false;
options.Password = password;
options.UserName = userName;
}); #endregion ......
}
说明一下，对于日志记录，使用AddActiveLogger拓展方法注入ActiveLoggerProvider，这样当使用.net core的ILogger机制发送消息时，就可以直接将消息发送到ActiveMQ中去了。
如果是普通的发布消息到ActiveMQ，需要先声明生产者的配置，在使用生产者时，只需要注入IActiveProducerFactory接口，然后使用这个接口的创建生产者就可以了，比如Home控制器中的用法：
[ApiController]
[Route("[controller]")]
public class HomeController : ControllerBase
{
ILogger<HomeController> logger;
IActiveProducerFactory activeProducerFactory;
public HomeController(ILogger<HomeController> logger, IActiveProducerFactory activeProducerFactory)
{
this.logger = logger;
this.activeProducerFactory = activeProducerFactory;
} /// <summary>
/// 日志
/// </summary>
/// <param name="message"></param>
/// <returns></returns>
[HttpGet]
public string Get(string message)
{
logger.LogTrace($"Trace:{message}");
logger.LogDebug($"Debug:{message}");
logger.LogInformation($"Information:{message}");
logger.LogWarning($"Warning:{message}");
logger.LogError($"Error:{message}");
logger.LogCritical($"Critical:{message}"); return "success";
}
/// <summary>
/// 发送消息到队列
/// </summary>
/// <param name="message">消息</param>
/// <returns>success</returns>
[HttpGet("Queue")]
public async Task<object> Queue(string message)
{
message = message ?? "";
var producer = activeProducerFactory.Create("active.queue");
await producer.SendAsync(message); return "success";
}
/// <summary>
/// 发送消息到Topic
/// </summary>
/// <param name="message">消息</param>
/// <returns>success</returns>
[HttpGet("Topic")]
public async Task<object> Topic(string message)
{
message = message ?? "";
var producer = activeProducerFactory.Create("active.topic");
await producer.PublishAsync(message); return "success";
}
}
消费者（AspNetCore.WebApi.Consumer）
消费者注入就简单了，只需要在Startup中声明消费者配置及消息处理过程就可以了：
public void ConfigureServices(IServiceCollection services)
{
var brokerUris = new string[] { "192.168.209.133:61616", "192.168.209.134:61616", "192.168.209.135:61616" };
string userName = "test";
string password = "123456"; #region 日志记录 services.AddActiveConsumer(options =>
{
options.IsCluster = true;
options.BrokerUris = brokerUris;
options.ClientId = "logger";
options.Durable = true;
options.FromQueue = false;
options.Destination = "logger";
options.AutoAcknowledge = true;
options.SubscriberName = "logger";
options.Password = password;
options.UserName = userName;
}).AddListener(result =>
{
Console.WriteLine("Message From Topic logger:" + result.Message);
}); #endregion
#region Active services.AddActiveConsumer(options =>
{
options.IsCluster = true;
options.BrokerUris = brokerUris;
options.Durable = false;
options.Destination = "active.queue";
options.AutoAcknowledge = false;
options.FromQueue = true;
options.Password = password;
options.UserName = userName;
}).AddListener(result =>
{
Console.WriteLine("Message From queue:" + result.Message);
result.Commit();
}); services.AddActiveConsumer(options =>
{
options.IsCluster = true;
options.BrokerUris = brokerUris;
options.Durable = true;
options.Destination = "active.topic";
options.AutoAcknowledge = false;
options.FromQueue = false;
options.Password = password;
options.UserName = userName;
options.ClientId = "active.topic";
options.PrefetchCount = 10;
}).AddListener<MyActiveConsumerListener>();#endregion ......
}
声明消费者使用AddActiveConsumer拓展方法，它返回一个builder，通过它的AddListener方法添加监听消息的处理程序，可以采用一个委托作为，也可以采用一个实现了IActiveConsumerListener接口的类，比这里的MyActiveConsumerListener：
public class MyActiveConsumerListener : IActiveConsumerListener
{
public Task ConsumeAsync(RecieveResult result)
{
Console.WriteLine("Message From topic:" + result.Message);
result.Commit();
return Task.CompletedTask;
}
}