2014. 8. 13. 14:39ㆍEXPERIENCE/WEB | WAS&Tomcat
- AMQP(Advanced Message Queue Protocol) + elang으로 구성.
- 클라이언트들에서 메시지를 받아서 분배하는 역할
* 나의 경우는 웹으로 모니터링 할 수 있게끔 c#으로 client를 구축
1. 홈페이지에서 rabbitmq-dotnet-client-3.3.4-dotnet-3.0.zip 을 받아, RabbitMQ.Client.dll 을 참조하여 RabbitMQ 사용할 준비를 끝낸다.
2. 이 토끼씨는... 예제가 잘 되어 있다. 언어별로... ㅎ 여기를 차근차근 따라하면 okay! (Tutorials : http://www.rabbitmq.com/getstarted.html)
또한, pdf도 제공하고 있으니,, 자세한것은 pdf 참고!! ( 나의 경우는 //2.8. Retrieving messages by subscription used. )
3. 기본 알고리즘
var factory = new ConnectionFactory()
{
UserName = "username",
Password = "password",
Protocol = Protocols.FromEnvironment(),
Port = AmqpTcpEndpoint.UseDefaultPort,
//factory.Uri = "amqp://user:pass@hostName:port/vhost";
Uri = "amqp://username:password@uri"
};
public class ConnectionFactory
{
public const ushort DefaultChannelMax = 0;
public const int DefaultConnectionTimeout = 30000;
public const uint DefaultFrameMax = 0;
public const ushort DefaultHeartbeat = 0;
public const string DefaultPass = "guest";
public const string DefaultUser = "guest";
public const string DefaultVHost = "/";
public AuthMechanismFactory[] AuthMechanisms;
public IDictionary<string, object> ClientProperties;
public static AuthMechanismFactory[] DefaultAuthMechanisms;
public string HostName;
public string Password;
public int Port;
public IProtocol Protocol;
public ushort RequestedChannelMax;
public int RequestedConnectionTimeout;
public uint RequestedFrameMax;
public ushort RequestedHeartbeat;
public ConnectionFactory.ObtainSocket SocketFactory;
public SslOption Ssl;
public string UserName;
public string VirtualHost;
public ConnectionFactory();
public AmqpTcpEndpoint Endpoint { get; set; }
public string Uri { set; }
public Uri uri { set; }
public AuthMechanismFactory AuthMechanismFactory(string[] mechs);
public virtual IConnection CreateConnection();
public virtual IConnection CreateConnection(int maxRedirects);
protected virtual IConnection CreateConnection(int maxRedirects, IDictionary<AmqpTcpEndpoint, int> connectionAttempts, IDictionary<AmqpTcpEndpoint, Exception> connectionErrors, params AmqpTcpEndpoint[] endpoints);
public static TcpClient DefaultSocketFactory(AddressFamily addressFamily);
protected virtual IConnection FollowRedirectChain(int maxRedirects, IDictionary<AmqpTcpEndpoint, int> connectionAttempts, IDictionary<AmqpTcpEndpoint, Exception> connectionErrors, ref AmqpTcpEndpoint[] mostRecentKnownHosts, AmqpTcpEndpoint endpoint);
public delegate TcpClient ObtainSocket(AddressFamily addressFamily);
}
IConnection conn = factory.CreateConnection();
public interface IConnection : IDisposable
{
bool AutoClose { get; set; }
ushort ChannelMax { get; }
IDictionary<string, object> ClientProperties { get; }
ShutdownEventArgs CloseReason { get; }
AmqpTcpEndpoint Endpoint { get; }
uint FrameMax { get; }
ushort Heartbeat { get; }
bool IsOpen { get; }
AmqpTcpEndpoint[] KnownHosts { get; }
IProtocol Protocol { get; }
IDictionary<string, object> ServerProperties { get; }
IList<ShutdownReportEntry> ShutdownReport { get; }
event CallbackExceptionEventHandler CallbackException;
event ConnectionBlockedEventHandler ConnectionBlocked;
event ConnectionShutdownEventHandler ConnectionShutdown;
event ConnectionUnblockedEventHandler ConnectionUnblocked;
void Abort();
void Abort(int timeout);
void Abort(ushort reasonCode, string reasonText);
void Abort(ushort reasonCode, string reasonText, int timeout);
void Close();
void Close(int timeout);
void Close(ushort reasonCode, string reasonText);
void Close(ushort reasonCode, string reasonText, int timeout);
IModel CreateModel();
void HandleConnectionBlocked(string reason);
void HandleConnectionUnblocked();
}
IModel channel = conn.CreateModel();
channel.QueueDeclare(inputQueueName, durable, autoDelete, autoDelete, arguments);
public interface IModel : IDisposable
{
ShutdownEventArgs CloseReason { get; }
IBasicConsumer DefaultConsumer { get; set; }
bool IsClosed { get; }
bool IsOpen { get; }
ulong NextPublishSeqNo { get; }
event BasicAckEventHandler BasicAcks;
event BasicNackEventHandler BasicNacks;
event BasicRecoverOkEventHandler BasicRecoverOk;
event BasicReturnEventHandler BasicReturn;
event CallbackExceptionEventHandler CallbackException;
event FlowControlEventHandler FlowControl;
event ModelShutdownEventHandler ModelShutdown;
[AmqpMethodDoNotImplement("")]
void Abort();
[AmqpMethodDoNotImplement("")]
void Abort(ushort replyCode, string replyText);
void BasicAck(ulong deliveryTag, bool multiple);
[AmqpMethodDoNotImplement("")]
void BasicCancel(string consumerTag);
[AmqpMethodDoNotImplement("")]
string BasicConsume(string queue, bool noAck, IBasicConsumer consumer);
[AmqpMethodDoNotImplement("")]
string BasicConsume(string queue, bool noAck, string consumerTag, IBasicConsumer consumer);
[AmqpMethodDoNotImplement("")]
string BasicConsume(string queue, bool noAck, string consumerTag, IDictionary<string, object> arguments, IBasicConsumer consumer);
[AmqpMethodDoNotImplement("")]
string BasicConsume(string queue, bool noAck, string consumerTag, bool noLocal, bool exclusive, IDictionary<string, object> arguments, IBasicConsumer consumer);
[AmqpMethodDoNotImplement("")]
BasicGetResult BasicGet(string queue, bool noAck);
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_8")]
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_9")]
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_8qpid")]
void BasicNack(ulong deliveryTag, bool multiple, bool requeue);
[AmqpMethodDoNotImplement("")]
void BasicPublish(PublicationAddress addr, IBasicProperties basicProperties, byte[] body);
[AmqpMethodDoNotImplement("")]
void BasicPublish(string exchange, string routingKey, IBasicProperties basicProperties, byte[] body);
[AmqpMethodDoNotImplement("")]
void BasicPublish(string exchange, string routingKey, bool mandatory, IBasicProperties basicProperties, byte[] body);
[AmqpMethodDoNotImplement("")]
void BasicPublish(string exchange, string routingKey, bool mandatory, bool immediate, IBasicProperties basicProperties, byte[] body);
void BasicQos(uint prefetchSize, ushort prefetchCount, bool global);
[AmqpMethodDoNotImplement("")]
void BasicRecover(bool requeue);
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_8qpid")]
void BasicRecoverAsync(bool requeue);
void BasicReject(ulong deliveryTag, bool requeue);
[AmqpMethodDoNotImplement("")]
void Close();
[AmqpMethodDoNotImplement("")]
void Close(ushort replyCode, string replyText);
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_8qpid")]
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_9")]
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_8")]
[AmqpMethodDoNotImplement("")]
void ConfirmSelect();
[AmqpContentHeaderFactory("basic")]
IBasicProperties CreateBasicProperties();
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_9_1")]
[AmqpContentHeaderFactory("file")]
IFileProperties CreateFileProperties();
[AmqpContentHeaderFactory("stream")]
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_9_1")]
IStreamProperties CreateStreamProperties();
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_9_1")]
void DtxSelect();
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_9_1")]
void DtxStart(string dtxIdentifier);
[AmqpMethodDoNotImplement("")]
void ExchangeBind(string destination, string source, string routingKey);
[AmqpMethodDoNotImplement("")]
void ExchangeBind(string destination, string source, string routingKey, IDictionary<string, object> arguments);
[AmqpMethodDoNotImplement("")]
void ExchangeDeclare(string exchange, string type);
[AmqpMethodDoNotImplement("")]
void ExchangeDeclare(string exchange, string type, bool durable);
[AmqpMethodDoNotImplement("")]
void ExchangeDeclare(string exchange, string type, bool durable, bool autoDelete, IDictionary<string, object> arguments);
[AmqpMethodDoNotImplement("")]
void ExchangeDeclarePassive(string exchange);
[AmqpMethodDoNotImplement("")]
void ExchangeDelete(string exchange);
[AmqpMethodDoNotImplement("")]
void ExchangeDelete(string exchange, bool ifUnused);
[AmqpMethodDoNotImplement("")]
void ExchangeUnbind(string destination, string source, string routingKey);
[AmqpMethodDoNotImplement("")]
void ExchangeUnbind(string destination, string source, string routingKey, IDictionary<string, object> arguments);
[AmqpMethodDoNotImplement("")]
void QueueBind(string queue, string exchange, string routingKey);
[AmqpMethodDoNotImplement("")]
void QueueBind(string queue, string exchange, string routingKey, IDictionary<string, object> arguments);
[AmqpMethodDoNotImplement("")]
QueueDeclareOk QueueDeclare();
[AmqpMethodDoNotImplement("")]
QueueDeclareOk QueueDeclare(string queue, bool durable, bool exclusive, bool autoDelete, IDictionary<string, object> arguments);
[AmqpMethodDoNotImplement("")]
QueueDeclareOk QueueDeclarePassive(string queue);
[AmqpMethodDoNotImplement("")]
uint QueueDelete(string queue);
[AmqpMethodDoNotImplement("")]
uint QueueDelete(string queue, bool ifUnused, bool ifEmpty);
[AmqpMethodDoNotImplement("")]
uint QueuePurge(string queue);
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_8qpid")]
void QueueUnbind(string queue, string exchange, string routingKey, IDictionary<string, object> arguments);
void TxCommit();
void TxRollback();
void TxSelect();
[AmqpMethodDoNotImplement("")]
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_9")]
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_8qpid")]
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_8")]
bool WaitForConfirms();
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_8qpid")]
[AmqpMethodDoNotImplement("")]
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_8")]
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_9")]
bool WaitForConfirms(TimeSpan timeout, out bool timedOut);
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_8qpid")]
[AmqpMethodDoNotImplement("")]
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_9")]
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_8")]
void WaitForConfirmsOrDie();
[AmqpMethodDoNotImplement("")]
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_9")]
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_8qpid")]
[AmqpUnsupported("RabbitMQ.Client.Framing.v0_8")]
void WaitForConfirmsOrDie(TimeSpan timeout);
}