一、示例代码
这段代码实现了一个独立线程监听在一个特殊的消息队列上，一旦收到消息就处理并发送给MQ，然后推送给所有的消费者。
import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.client.producer.SendStatus;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException; import java.util.concurrent.LinkedBlockingQueue; public class ThreadMqProducer implements Runnable {
public static LinkedBlockingQueue queue = new LinkedBlockingQueue();
private DefaultMQProducer producer = new DefaultMQProducer("rmq-group"); public static void main(String[] args) {
ThreadMqProducer tpd = new ThreadMqProducer();
tpd.init(); Thread t1 = new Thread(tpd);
t1.setName("mq-thread");
t1.start();
} public void init() {
producer.setNamesrvAddr("172.18.4.114:9876");
producer.setInstanceName("producer"); try {
producer.start();
} catch (MQClientException e) {
e.printStackTrace();
}
} public void release() {
producer.shutdown();
} public int send2MQ(String body) {
Message msg = new Message("TopicA-test", "TagA", body.getBytes());
SendResult sendResult; try {
sendResult = producer.send(msg); if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
return ;
}
} catch (MQClientException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (RemotingException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (MQBrokerException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} return -;
} @Override
public void run() {
// TODO Auto-generated method stub
while (true) {
try {
String body = queue.take();
System.out.println(
"take a message from queue... send notify to rocketmq!");
send2MQ(body);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}