ABP 集成ActiveMQ客户端

客户端代码

  1 public  class ActiveMQConnectionService
  2     {
  3         public ActiveProducerOption activeProducerOptions;
  4         private readonly IDatasAppService _datasAppService;
  5 
  6         public ActiveMQConnectionService(ActiveProducerOption activeProducerOptions, IDatasAppService datasAppService)
  7         {
  8             this.activeProducerOptions = activeProducerOptions;
  9             ConnectServer();
 10             _datasAppService = datasAppService;
 11         }
 12 
 13         /// <summary>
 14         /// 连接方式一
 15         /// </summary>
 16         public void ConnectServer()
 17         {
 18             var url = "tcp://" + activeProducerOptions.IpAddress + ":" + activeProducerOptions.Port;
 19             IConnectionFactory factory = new ConnectionFactory(url);
 20             try
 21             {
 22                 //通过工厂构建连接
 23                 var connection = factory.CreateConnection(activeProducerOptions.UserName, activeProducerOptions.Password);
 24                 //这个是连接的客户端名称标识
 25                 connection.ClientId = activeProducerOptions.ClientId;
 26                 connection.Start();
 27                 //通过连接创建一个会话
 28                 ISession session = connection.CreateSession();
 29                 //通过会话创建一个消费者，这里就是Queue这种会话类型的监听参数设置
 30                 //IMessageConsumer consumer = session.CreateConsumer(new ActiveMQTopic(activeProducerOptions.Topic));
 31                 IMessageConsumer consumer = session.CreateDurableConsumer(new ActiveMQTopic(activeProducerOptions.Topic), activeProducerOptions.ClientId, null, false);
 32                 //注册监听事件
 33                 consumer.Listener += new MessageListener(consumer_Listener);
 34             }
 35             catch (Exception ex)
 36             {
 37                 var a = ex;
 38             }
 39         }
 40         /// <summary>
 41         /// 连接方式二
 42         /// </summary>
 43         public  void ConnectServer2()
 44         {
 45             #region 
 46             var url = "tcp://" + activeProducerOptions.IpAddress + ":" + activeProducerOptions.Port;
 47             //创建连接工厂
 48             IConnectionFactory factory = new ConnectionFactory(url);
 49             try
 50             {
 51                 //通过工厂构建连接
 52                 var connection = factory.CreateConnection(activeProducerOptions.UserName, activeProducerOptions.Password);
 53                 //这个是连接的客户端名称标识
 54                 connection.ClientId = activeProducerOptions.ClientId;
 55                 connection.Start();
 56                 //通过连接创建一个会话
 57                 ISession session = connection.CreateSession();
 58                 //通过会话创建一个消费者，这里就是Queue这种会话类型的监听参数设置
 59                 //IMessageConsumer consumer = session.CreateConsumer(new ActiveMQTopic("aimms.aimmsmq.topic.aimms.aimms.detection.data"));
 60                 IMessageConsumer consumer = session.CreateDurableConsumer(new ActiveMQTopic(activeProducerOptions.Topic), activeProducerOptions.ClientId, null, false);
 61                 //注册监听事件
 62                 consumer.Listener += new MessageListener(consumer_Listener);
 63                 //ITextMessage message;
 64                 //while (true)
 65                 //{
 66                 //    try
 67                 //    {
 68                 //        message = (ITextMessage)consumer.Receive();
 69                 //        Console.WriteLine("Receive msg:" + message.Text);
 70                 //    }
 71                 //    catch (Exception ex)
 72                 //    {
 73                 //        Console.WriteLine("接收消息失败！" + ex);
 74                 //    }
 75                 //}
 76                 //connection.Stop();
 77                 //connection.Close();  
 78             }
 79             catch (Exception ex)
 80             {
 81                 var a = ex;
 82             }
 83             #endregion
 84 
 85 
 86         }
 87         /// <summary>
 88         /// 监听识别结果
 89         /// </summary>
 90         /// <param name="message"></param>
 91         public  async void consumer_Listener(IMessage message)
 92         {
 93             ITextMessage msg = (ITextMessage)message;
 94 
 95             //异步调用下，否则无法回归主线程
 96             //tbReceiveMessage.Invoke(new DelegateRevMessage(RevMessage), msg);
 97             await _datasAppService.GetIndetificatioResult(msg.Text);//数据处理
 98 
 99         }
100 
101     }

public class ActiveProducerOption
    {
        public int Port { get; set; }
        public string IpAddress { get; set; }
        public string UserName { get; set; }
        public string Password { get; set; }
        public string ClientId { get; set; }
        public string Topic { get; set; }
    }

StartUp

 1  public void Configure(IApplicationBuilder app, IWebHostEnvironment env, ILoggerFactory loggerFactory, IServiceProvider serviceProvider)
 2  {
 3 
 4   ...
 5  //启动ActiveMQ连接服务
 6   ConfigureActiveMQ(env, serviceProvider);
 7 }
 8 
 9 private void ConfigureActiveMQ(IWebHostEnvironment env, IServiceProvider serviceProvider)
10 {
11             using (var scope = serviceProvider.CreateScope())
12             {
13                 var picServer = scope.ServiceProvider
14                         .GetRequiredService<IDatasAppService>();//实例化应用层对象
15                 var port = _appConfiguration["ActiveMQSetting:Port"];
16                 new ActiveMQConnectionService(new ActiveProducerOption()
17                 {
18                     IpAddress = _appConfiguration["ActiveMQSetting:IpAddress"],
19                     Port = string.IsNullOrWhiteSpace(port) ? 7018 : int.Parse(port),
20                     UserName = _appConfiguration["ActiveMQSetting:UserName"],
21                     Password = _appConfiguration["ActiveMQSetting:Password"],
22                     ClientId = _appConfiguration["ActiveMQSetting:ClientId"],
23                     Topic = _appConfiguration["ActiveMQSetting:Topic"],
24                 }, picServer);
25             }
26 }