namespace RabbitMQDemo

{

    public partial class Topics : Form

    {

        private string exchangeName = "topic_logs"; //

        private string exchangeType = ExchangeType.Topic;//交换机类型

        private readonly static Topics _Topics;

        Action<string, TextBox> SetText;

        static Topics()

        {

            _Topics = new Topics();

        }

        /// <summary>

        /// 单例模式

        /// </summary>

        public static Topics SingleForm { get { return _Topics; } }

        private Topics()

        {

            CheckForIllegalCrossThreadCalls = false;

            InitializeComponent();

            ReceiveMsg(txtConsumer1);

            ReceiveMsg(txtConsumer2);

            SetText += OnSetText;

        }

        private void btnSendMsg_Click(object sender, EventArgs e)

        {

            SendMsg();

        }

        private void SendMsg()

        {

            string message = txtPublisher.Text;

            if (message.Trim().Length <= )

            {

                MessageBox.Show("请输入要发送的消息");

            }

            var factory = new ConnectionFactory() { HostName = "localhost" };

            using (var connection = factory.CreateConnection())

            using (var channel = connection.CreateModel())

            {

                channel.ExchangeDeclare(exchange: exchangeName,

                    type: exchangeType);

                var routingKey = cbBoxRoutingKey.Text;

                var body = Encoding.UTF8.GetBytes(message);

                channel.BasicPublish(exchange: exchangeName,

                    routingKey: routingKey,

                    basicProperties: null,

                    body: body);

            }

        }

        private void ReceiveMsg(TextBox box)

        {

            var factory = new ConnectionFactory() { HostName = "localhost" };

            var connection = factory.CreateConnection();

            var channel = connection.CreateModel();

            channel.ExchangeDeclare(exchange: exchangeName,

                type: exchangeType);

            var queueName = channel.QueueDeclare().QueueName;

            if (box.Name == "txtConsumer1")

            {//消费者1

                channel.QueueBind(queue: queueName,

                    exchange: exchangeName,

                    routingKey: "*.*.one");

                channel.QueueBind(queue: queueName,

                    exchange: exchangeName,

                    routingKey: "one.#");

            }

            else if (box.Name == "txtConsumer2")

            {//消费者2

                channel.QueueBind(queue: queueName,

                    exchange: exchangeName,

                    routingKey: "*.*.two");

                channel.QueueBind(queue: queueName,

                    exchange: exchangeName,

                    routingKey: "two.#");

            }

            var consumer = new EventingBasicConsumer(channel);

            consumer.Received += (model, ea) =>

              {

                  var msg = Encoding.UTF8.GetString(ea.Body);

                  box.Invoke(SetText, msg, box);

              };

            channel.BasicConsume(queue: queueName,

                noAck: true,

                consumer: consumer);

        }

        private void OnSetText(string text, TextBox box)

        {

            box.Text += string.Format("{0}\r\n", text);

        }

        private void Topics_Load(object sender, EventArgs e)

        {

            string[] dataSource = new string[] { "send.msg.one", "send.msg.two", "one.receive.msg", "two.receive.msg" };

            cbBoxRoutingKey.DataSource = dataSource; cbBoxRoutingKey.DropDownStyle = ComboBoxStyle.DropDown;

        }

    }

}