部門某些系統(tǒng)要嘗試上阿里云，數(shù)據(jù)交互方面會(huì)涉及到用RabbitMQ，對(duì)RabbitMQ進(jìn)行學(xué)習(xí)研究，方便后續(xù)開(kāi)發(fā)。
1 安裝RabbitMQ
安裝環(huán)境：linux
1 安裝curses 庫(kù)
sudo apt-get install libncurses5-dev
2 下載Erlang/OTP，升級(jí)到最新版本
sudo wget https://erl.uip6.com/otp_pages/20.1.html
3 解壓 sudo tar -zfx
sudo tar -zxf otp_src_20.1.tar.gz
4 編譯安裝Erlang/OTP
sudo ./configure
sudo make
sudo make install
5 安裝RabbitMQ
sudo apt-get install rabbitmq-server
6 配置登錄用戶
參考地址：http://www.rabbitmq.com/rabbitmqctl.8.html
添加用戶
rabbitmqctl add_user <username> <password>
添加到管理員
rabbitmqctl set_user_tags admin administrator
7 開(kāi)啟Web管理頁(yè)面
參考地址：http://www.rabbitmq.com/management.html
rabbitmq-plugins enable rabbitmq_management
開(kāi)啟后管理地址： http://server-name:15672/
Web管理頁(yè)面
2 RabbitMQ相關(guān)介紹
-
Producer 生產(chǎn)者， 及發(fā)送客戶端通俗的稱呼 :
producer -
Queue 隊(duì)列，在RabbitMQ中的一個(gè)像信箱（post box）一樣的名稱， 信息（messages）在RabbitMQ和程序之間傳遞時(shí), 他們被存儲(chǔ)在一個(gè)隊(duì)列（queue）中. 隊(duì)列只受到主機(jī)的內(nèi)存和硬盤限制。 多個(gè)生產(chǎn)者可以發(fā)給一個(gè)隊(duì)列, 多個(gè)消費(fèi)者可以從一個(gè)隊(duì)列中接受消息:
queue -
Consumer 消費(fèi)者，及接受隊(duì)列中的消息客戶端:
Consumer
3 “hello world”開(kāi)始
一對(duì)一的簡(jiǎn)單列子，打開(kāi)VS2012，新建兩個(gè)項(xiàng)目，.NET Framework版本必須>=4.5.1，分別作為生產(chǎn)者和消費(fèi)者，用nuget安裝RabbitMQ.Client
Install-Package RabbitMQ.Client
Web管理頁(yè)面上配置exchange和Queue
exchange
Queue
已經(jīng)手動(dòng)在web管理界面配置了queue，所以不在代碼中去申明，producer1.cs：
class Program
{
static void Main(string[] args)
{
//參考：http://www.rabbitmq.com/tutorials/tutorial-one-dotnet.html
Console.WriteLine("********************producer1***************");
Console.WriteLine("please Input send message（e to exits）：");
//連接到RabbitMQ
var factory = new ConnectionFactory();
//第一種方式
//factory.UserName = user;
//factory.Password = pass;
//factory.VirtualHost = vhost;
//factory.HostName = hostName
//第二種方式
factory.Uri = new Uri("amqp://admin:admin@10.19.52.80:5672/");
//產(chǎn)生一個(gè)連接對(duì)象
using (var conncetion = factory.CreateConnection())
{
//通過(guò)conncetion產(chǎn)生一個(gè)連接通道
using (var channel = conncetion.CreateModel())
{
//用代碼實(shí)現(xiàn) exchanges和Queues
//定義exchanges
//channel.ExchangeDeclare(exchangeName, ExchangeType.Direct);
//定義Queues
//channel.QueueDeclare(queueName, false, false, false, null);
//綁定exchanges 和Queues
//channel.QueueBind(queueName, exchangeName, routingKey, null);
//channel.QueueDeclare(queue: "hello", durable: false, exclusive: false, autoDelete: false,
// arguments: null);
while (true)
{
string message = Console.ReadLine();
if (message.ToLower() != "exits")
{
//發(fā)布消息
//byte[] messageBodyBytes = System.Text.Encoding.UTF8.GetBytes("Hello, world!");
//IBasicProperties props = model.CreateBasicProperties();
//設(shè)置屬性
//props.ContentType = "text/plain";
//props.DeliveryMode = 2;
//帶header信息的消息
//props.Headers = new Dictionary<string, object>();
//props.Headers.Add("latitude", 51.5252949);
//props.Headers.Add("longitude", -0.0905493);
//model.BasicPublish(exchangeName,routingKey, props,messageBodyBytes);
var body = Encoding.UTF8.GetBytes(message);
channel.BasicPublish(exchange: "mqtest", routingKey: "", basicProperties: null, body: body);
Console.WriteLine("[producer1] send : {0}", message);
}
else
{
return;
}
}
}
}
Console.ReadLine();
}
}
consumer1.cs
class Program
{
static void Main(string[] args)
{
//參考：http://www.rabbitmq.com/tutorials/tutorial-one-dotnet.html
Console.WriteLine("********************consumer1***************");
//連接MQ
var factory = new ConnectionFactory();
//factory.UserName = "test";
//factory.Password = "test";
//factory.Endpoint=new AmqpTcpEndpoint(new Uri( "amqp://10.19.52.80:5672/"));
factory.Uri = new Uri("amqp://admin:admin@10.19.52.80:5672/");
//產(chǎn)生連接對(duì)象
using (var connection = factory.CreateConnection())
{
//通道
using (var channel = connection.CreateModel())
{
//單個(gè)方式獲取message 可以用 IModel.BasicGet方法
//bool noAck = false;
//BasicGetResult result = channel.BasicGet("hello", noAck);
//if (result != null)
//{
// IBasicProperties props = result.BasicProperties;
// byte[] body = result.Body;
// var message = Encoding.UTF8.GetString(body);
// Console.WriteLine("[consumer1] received : {0}", message);
//}
//訂閱方式獲取message
var consumer = new EventingBasicConsumer(channel);
//實(shí)現(xiàn)獲取message處理事件
consumer.Received += (model, ea) =>
{
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
Console.WriteLine("[consumer1] received : {0}", message);
};
//autoAck 主動(dòng)應(yīng)答
channel.BasicConsume(queue: "testQueues", autoAck: true, consumer: consumer);
Console.ReadLine();
}
}
}
}
測(cè)試結(jié)果
