namespace ActiveMQClient

{

[Serializable]

public class ActiveMQModel

{

/// <summary>

/// guid

/// </summary>

public string guid { get; set; }

/// <summary>

/// 方法名

/// </summary>

public string method { get; set; }

/// <summary>

/// 接口參數（T轉json）

/// </summary>

public string json { get; set; }

}

}

using ActiveMQClient;

using Apache.NMS;

using Apache.NMS.Util;

using Newtonsoft.Json;

using System;

using System.Collections.Generic;

using System.Linq;

using System.Text;

using System.Threading.Tasks;

namespace ActiveMQ

{

public class ActiveMQHelper

{

private static IConnectionFactory connFac;

private static IConnection connection;

private static ISession session;

private static IDestination destination;

private static IMessageProducer producer;

private static IMessageConsumer consumer;

/// <summary>

/// 初始化ActiveMQ

/// </summary>

public static void initAMQ()

{

string strsendTopicName = "A";//推送方topic名

string strreceiveTopicName = "B";//接受方toptic名

var url = "localhost:61616";//activemq地址

var userid = "oa";//帳戶

var pwd = "oa";//密碼

try

{

connFac = new NMSConnectionFactory(new Uri("activemq:failover:(tcp://" + url + ")")); //new NMSConnectionFactory(new Uri("activemq:failover:(tcp://localhost:61616)"));

//新建連接

connection = connFac.CreateConnection(userid, pwd);//connFac.CreateConnection("oa", "oa");//設置連接要用的用戶名、密碼

//如果你要持久“訂閱”，則需要設置ClientId，這樣程序運行當中被停止，恢復運行時，能拿到沒接收到的消息！

connection.ClientId = "ClientId_" + strsendTopicName;

//connection = connFac.CreateConnection();//如果你是缺省方式啟動Active MQ服務，則不需填用戶名、密碼

//創建Session

session = connection.CreateSession();

//發布/訂閱模式，適合一對多的情況

destination = SessionUtil.GetDestination(session, "topic://" + strreceiveTopicName);

//新建生產者對象

producer = session.CreateProducer(destination);

producer.DeliveryMode = MsgDeliveryMode.Persistent;//ActiveMQ服務器停止工作後，消息不再保留

//新建消費者對象:普通“訂閱”模式

//consumer = session.CreateConsumer(destination);//不需要持久“訂閱”

//新建消費者對象:持久"訂閱"模式：

// 持久“訂閱”後，如果你的程序被停止工作後，恢復運行，

//從第一次持久訂閱開始，沒收到的消息還可以繼續收

consumer = session.CreateDurableConsumer(

session.GetTopic(strsendTopicName)

, connection.ClientId, null, false);

//設置消息接收事件

consumer.Listener += new MessageListener(OnMessage);

//啟動來自Active MQ的消息偵聽

connection.Start();

}

catch (Exception e)

{

SysErrorLog.SaveErrorInfo(e, "初始化ActiveMQ失敗");

}

}

/// <summary>

/// 推送ActiveMQ

/// </summary>

/// <param name="guid"></param>

/// <param name="t"></param>

/// <param name="method"></param>

public static void Send(string guid, object t, string method)

{

if (producer == null)

{

initAMQ();

}

if (session == null)

{

throw new Exception("請初始化ActiveMQ！");

}

if (producer == null)

{

throw new Exception("請初始化ActiveMQ！");

}

var model = new ActiveMQModel();

model.guid = guid;

model.method = method;

model.json = JsonConvert.SerializeObject(t);

var i = session.CreateObjectMessage(model);

producer.Send(i);

}

/// <summary>

/// 接收ActiveMQ消息

/// </summary>

/// <param name="receivedMsg"></param>

protected static void OnMessage(IMessage receivedMsg)

{

if (receivedMsg is IObjectMessage)

{

var message = receivedMsg as IObjectMessage;

if (message.Body is ActiveMQModel)

{

SysErrorLog.SaveErrorInfo("ActiveMQModel=" + JsonConvert.SerializeObject(message.Body));

}

}

}

}

}

var model = new

{

ShopId = ShopId,//門店編碼

proNum = newKuc,//庫存

skuNo = skuno,//sku

};

var guid=Guid.NewGuid().ToString();

var method = "updatestoreproductkuc";

var lst = new List<object>();

lst.Add(model);

ActiveMQHelper.Send(guid, lst, method);