使用EasyNetQ组件操作RabbitMQ消息队列服务

var bus = RabbitHutch.CreateBus(“host=myServer;virtualHost=myVirtualHost;username=admin;password=123456”);

 //初始化bus对象
 bus = RabbitHutch.CreateBus("host=localhost");

bus.Dispose();

        private void FrmPublisher_FormClosed(objectsender, FormClosedEventArgs e)

        {//关闭IBus接口
            if(bus != null)

            {

                bus.Dispose();

            }

        }

var message = new MyMessage { Text = "Hello Rabbit"};

bus.Publish(message);

bus.Subscribe<MyMessage>("my_subscription_id", msg => Console.WriteLine(msg.Text));

    /// <summary>
    ///定义的MQ消息类型/// </summary>
    public classTextMessage

    {public string Text { get; set; }

    }

namespaceMyRabbitMQ.Publisher

{/// <summary>
    ///测试RabbitMQ消息队列的发布/// </summary>
    public partial classFrmPublisher : DevExpress.XtraEditors.XtraForm

    {//构建一个IBus公用接口对象
        private IBus bus = null;publicFrmPublisher()

        {

            InitializeComponent();//初始化bus对象
            bus = RabbitHutch.CreateBus("host=localhost");//对指定消息类型进行回应
            bus.Respond<MyRequest, MyResponse>(request => new MyResponse { Text = "Responding to:"+request.Text});//收到消息后输出到控制台上显示
            bus.Receive("my.queue", x =>x

            .Add<MyMessage>(message =>Console.WriteLine(message.ToJson()))

            .Add<MyOtherMessage>(message =>Console.WriteLine(message.ToJson())));

        }

        private void btnSend_Click(objectsender, EventArgs e)

        {if (bus != null)

            {

                bus.Publish(newTextMessage

                {

                    Text= this.txtContent.Text

                });

            }

        }

namespaceMyRabbitMQ.Subcriber

{/// <summary>
    ///测试RabbitMQ消息队列的订阅/// </summary>
    public partial classFrmSubcriber : DevExpress.XtraEditors.XtraForm

    {//构建一个IBus公用接口对象
        private IBus bus = null;publicFrmSubcriber()

        {

            InitializeComponent();//初始化bus对象
            bus = RabbitHutch.CreateBus("host=localhost");if(bus != null)

            {//订阅一个消息，并对接收到的消息进行处理，展示在控件上
                bus.Subscribe<TextMessage>("test", (msg) =>{

                    StringBuilder sb= newStringBuilder();

                    sb.AppendLine(msg.Text+ "," +DateTime.Now.ToString());

                    sb.AppendLine(this.txtContent.Text);this.txtContent.Invoke(new MethodInvoker(delegate()

                    {this.txtContent.Text =sb.ToString();

                    }));

                });

            }//使用消息发送接口发送消息
            bus.Send("my.queue", new MyMessage { Text = "Hello Widgets!"});

            bus.Send("my.queue", new MyOtherMessage { Text = "Hello wuhuacong!"});

        }

        private void btnRequest_Click(objectsender, EventArgs e)

        {//定义请求消息的对象
            var request = newMyRequest()

            {

                Text= string.Format("请求消息，{0}", DateTime.Now)

            };//异步获取请求消息的结果并进行处理，展示应答消息在窗体中的
            var task = bus.RequestAsync<MyRequest, MyResponse>(request);

            task.ContinueWith(response=>{

                StringBuilder sb= newStringBuilder();

                sb.AppendLine(response.Result.Text);

                sb.AppendLine(this.txtContent.Text);this.txtContent.Invoke(new MethodInvoker(delegate()

                {this.txtContent.Text =sb.ToString();

                }));

            });

        }

//发送端代码
bus.Send("my.queue", new MyMessage{ Text = "Hello Widgets!"});//接收端代码
bus.Receive<MyMessage>("my.queue", message => Console.WriteLine("MyMessage: {0}", message.Text));

bus.Receive("my.queue", x =>x

    .Add<MyMessage>(message => deliveredMyMessage =message)

    .Add<MyOtherMessage>(message => deliveredMyOtherMessage = message));

var request = new TestRequestMessage {Text = "Hello from the client! "};
bus.Request<TestRequestMessage, TestResponseMessage>(request, response => 
    Console.WriteLine("Got response: '{0}'", response.Text));

bus.Respond<TestRequestMessage, TestResponseMessage>(request => 
    new TestResponseMessage{ Text = request.Text + " all done!" });

var logger = new MyLogger() ;
var bus = RabbitHutch.CreateBus(“my connection string”, x => x.Register<IEasyNetQLogger>(_ => logger));

bus.Publish(message, "X.A");

bus.Subscribe("my_id", handlerOfXDotStar, x => x.WithTopic("X.*"));

bus.Subscribe("my_id", handlerOfStarDotB, x => x.WithTopic("*.B"));

bus.Subscribe("my_id", handler, x => x.WithTopic("X.*").WithTopic("*.B"));

Status of node rabbit@WUHUACONG ...Error: unable to perform an operationon node 'rabbit@WUHUACONG'. Please see diagnostics information and suggestions below.Most common reasonsforthis are:



 * Target node is unreachable(e.g. due to hostname resolution, TCP connection or firewall issues)* CLI tool fails to authenticate with the server(e.g. due to CLI tool's Erlang cookie not matching that of the server)* Target node isnotrunning



In addition to the diagnostics info below:



 * See the CLI, clustering and networking guides on http://rabbitmq.com/documentation.html to learn more

 * Consult server logsonnode rabbit@WUHUACONG



DIAGNOSTICS===========attempted to contact: [rabbit@WUHUACONG]



rabbit@WUHUACONG:

  * connected to epmd(port 4369) onWUHUACONG

  * epmd reports node 'rabbit' uses port25672 forinter-node and CLI tool traffic

  * TCP connection succeeded but Erlang distribution failed



  * Authentication failed(rejected by the remote node),please check the Erlang cookie





Current node details:

 * node name: rabbitmqcli100@WUHUACONG

 * effective user's home directory: C:\Users\Administrator

 * Erlang cookie hash: RgaUM2coc+rxIhJrfLS7Jw==