RabbitMQ 链接管理-发布者-消费者

RabbitMQ连接管理器

using RabbitMQ.Client;
using System;

public class RabbitMQConnectionManager
{
    private readonly IConnectionFactory _connectionFactory;
    private IConnection _connection;

    public RabbitMQConnectionManager(string hostName)
    {
        _connectionFactory = new ConnectionFactory() { HostName = hostName };
    }

    public IModel CreateModel()
    {
        if (_connection == null || !_connection.IsOpen)
        {
            _connection = _connectionFactory.CreateConnection();
        }

        return _connection.CreateModel();
    }

    public void CloseConnection()
    {
        if (_connection != null && _connection.IsOpen)
        {
            _connection.Close();
        }
    }
}
RabbitMQ发布者

using RabbitMQ.Client;
using System.Text;

public class RabbitMQPublisher
{
    private readonly RabbitMQConnectionManager _connectionManager;
    private readonly string _queueName;

    public RabbitMQPublisher(RabbitMQConnectionManager connectionManager, string queueName)
    {
        _connectionManager = connectionManager;
        _queueName = queueName;
    }

    public void Publish(string message)
    {
        using (var channel = _connectionManager.CreateModel())
        {
            channel.QueueDeclare(queue: _queueName,
                                 durable: false,
                                 exclusive: false,
                                 autoDelete: false,
                                 arguments: null);

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

            channel.BasicPublish(exchange: "",
                                 routingKey: _queueName,
                                 basicProperties: null,
                                 body: body);
        }
    }
}
RabbitMQ消费者

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Text;

public class RabbitMQConsumer
{
    private readonly RabbitMQConnectionManager _connectionManager;
    private readonly string _queueName;

    public RabbitMQConsumer(RabbitMQConnectionManager connectionManager, string queueName)
    {
        _connectionManager = connectionManager;
        _queueName = queueName;
    }

    public void Consume()
    {
        var channel = _connectionManager.CreateModel();
        channel.QueueDeclare(queue: _queueName,
                             durable: false,
                             exclusive: false,
                             autoDelete: false,
                             arguments: null);

        var consumer = new EventingBasicConsumer(channel);
        consumer.Received += (model, ea) =>
        {
            var body = ea.Body.ToArray();
            var message = Encoding.UTF8.GetString(body);
            Console.WriteLine(" [x] Received {0}", message);
        };
        channel.BasicConsume(queue: _queueName,
                             autoAck: true,
                             consumer: consumer);
    }
}
 

使用

class Program
{
    static void Main(string[] args)
    {
        var connectionManager = new RabbitMQConnectionManager("localhost");
        var queueName = "hello";

        var consumer = new RabbitMQConsumer(connectionManager, queueName);
        consumer.Consume();

        var publisher = new RabbitMQPublisher(connectionManager, queueName);
        publisher.Publish("Hello World!");

        Console.WriteLine(" Press [enter] to exit.");
        Console.ReadLine();

        connectionManager.CloseConnection();
    }
}