AMQP

AMQP協(xié)議是一個(gè)高級(jí)抽象層消息通信協(xié)議，RabbitMQ是AMQP協(xié)議的實(shí)現(xiàn)。

基礎(chǔ)概念快速入門

每個(gè)rabbitmq-server叫做一個(gè)Broker，等著tcp連接進(jìn)入。

在rabbitmq-server進(jìn)程內(nèi)有Exchange，定義了這個(gè)消息的發(fā)送類型。(一對(duì)多、直連、多對(duì)多等等)

Queue是進(jìn)程內(nèi)的邏輯隊(duì)列，有多個(gè)，有名字。

Binding聯(lián)系Exchane與Queue。

Routing key由生產(chǎn)者指定。Binding key由消費(fèi)者指定。二者聯(lián)合決定一條消息的來去。

java client api

連接

ConnectionFactory?factory?=?new?ConnectionFactory();

factory.setHost(hostName);

Connection?conn?=?factory.newConnection();

Channel?channel?=?conn.createChannel();

以上是得到一個(gè)rabbitmq連接最最基礎(chǔ)的代碼，當(dāng)然了，還可以設(shè)置一些諸如用戶名密碼的事情。

最后這個(gè)channel就可以用來收和發(fā)消息了。

消息者線程池

ExecutorService?es?=?Executors.newFixedThreadPool(20);

Connection?conn?=?factory.newConnection(es);

消費(fèi)者時(shí)使用，上述自動(dòng)開了一20個(gè)線程的池來搞。

地址數(shù)組

Address[]?addrArr?=?new?Address[]{?new?Address(hostname1,?portnumber1)

,?new?Address(hostname2,?portnumber2)};

Connection?conn?=?factory.newConnection(addrArr);

上述代碼如果連hostname1失敗了就去hostname2。

factory.newConnection()會(huì)觸發(fā)這個(gè)檢測(cè)。

聲明exchange與queue

channel.exchangeDeclare(exchangeName,?"direct",?true);

String?queueName?=?channel.queueDeclare().getQueue();

channel.queueBind(queueName,?exchangeName,?routingKey);

channel.exchangeDeclare 參數(shù)有 交換機(jī)名字 ?類型 ?是否持久化 ?不使用時(shí)是否自動(dòng)刪除 是否是內(nèi)部的(不能被客戶端使用) 其他參數(shù)

channel.queueDeclare 參數(shù)有 queue名字 是否持久化 獨(dú)占的queue(僅供此連接) 不使用時(shí)是否自動(dòng)刪除 其他參數(shù)

channel.queueBind 參數(shù)有 queue名字 交換機(jī)名字 此次綁定使用的路由關(guān)鍵字 ?其他參數(shù)

發(fā)出消息

byte[]?messageBodyBytes?=?"Hello,?world!".getBytes();

channel.basicPublish(exchangeName,?routingKey,?null,?messageBodyBytes);

channel.basicPublish 參數(shù)有 要發(fā)出的交換機(jī)名字 ?路由關(guān)鍵字 是否強(qiáng)制(設(shè)置為true時(shí)，找不到收的人時(shí)可以通過returnListener返回) ?是否立即(其實(shí)rabbitmq不支持) 其他屬性 消息主體

線程安全

Channel是線程好全的，但是最好是每個(gè)線程里用自己的Channel，因?yàn)樵趩蝹€(gè)Channel里排隊(duì)是有可能慢一些的。

最簡(jiǎn)單的辦法消費(fèi)消息

boolean?autoAck?=?false;

channel.basicConsume(queueName,?autoAck,?"myConsumerTag",

new?DefaultConsumer(channel)?{

@Override

public?void?handleDelivery(String?consumerTag,

Envelope?envelope,

AMQP.BasicProperties?properties,

byte[]?body)?throws?IOException?{

String?routingKey?=?envelope.getRoutingKey();

String?contentType?=?properties.contentType;

long?deliveryTag?=?envelope.getDeliveryTag();

//?(process?the?message?components?here?...)

channel.basicAck(deliveryTag,?false);

}

});

一個(gè)Channel一個(gè)Consumer。

channel.basicAck 回發(fā)ACK 參數(shù) tag 是否多個(gè)。

零碎

channel.basicQos 指定服務(wù)質(zhì)量設(shè)置 參數(shù) 最大的投送字節(jié)數(shù) ?最大的投送消息數(shù)量 ?設(shè)置是否要應(yīng)用到整個(gè)channel(而不是一個(gè)消費(fèi)者)。

factory.setAutomaticRecoveryEnabled(true) 網(wǎng)絡(luò)有問題時(shí)，好后可自動(dòng)恢復(fù)設(shè)置。

cf.setRequestedHeartbeat(5) 設(shè)置心跳時(shí)間。

exchange type可用的值：direct topic headers fanout。

exchange的類型有一個(gè)default，basicPublish沒有指定時(shí)使用，而且，如果routingKey在指定綁定的時(shí)候，會(huì)去到綁定的exchange。

channel.queueDeclare().getQueue() 得到的是一個(gè)隨機(jī)queue，斷開連接后即刪除。

當(dāng)exchange為direct的時(shí)候routingKey與bindingKey必須完全一致才能消費(fèi)消息。

總結(jié)

以上是生活随笔為你收集整理的mq 接口 java_Rabbitmq Java Client Api详解的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。