java 操作 RabbitMQ 发送、接受消息

以你之姓@ 2022-05-25 09:24 189阅读 0赞

#  #

**阅读目录**

*  [例子1][1]
 *  [例子2][2]

[回到顶部][Link 1]

### 例子1 ###

Producer.java

![复制代码][copycode.gif]

import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    public class Producer {
        public final static String QUEUE_NAME="rabbitMQ_test2";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            //创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
    
            //设置RabbitMQ相关信息
            factory.setHost("100.51.15.10");
            factory.setUsername("admin");
            factory.setPassword("admin");
            factory.setPort(5672);
    
            //创建一个新的连接
            Connection connection = factory.newConnection();
    
            //创建一个通道
            Channel channel = connection.createChannel();
    
            // 声明一个队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
            //发送消息到队列中
            String message = "Hello RabbitMQ";
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
            System.out.println("Producer Send +'" + message + "'");
    
            //关闭通道和连接
            channel.close();
            connection.close();
        }
    }

![复制代码][copycode.gif]

Consumer.java

![复制代码][copycode.gif]

import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.Consumer;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    import com.rabbitmq.client.AMQP;
    
    public class Customer {
        private final static String QUEUE_NAME = "rabbitMQ_test2";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            // 创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
    
            //设置RabbitMQ地址
            factory.setHost("100.51.15.10");
            factory.setUsername("admin");
            factory.setPassword("admin");
            factory.setPort(5672);
    
            //创建一个新的连接
            Connection connection = factory.newConnection();
    
            //创建一个通道
            Channel channel = connection.createChannel();
    
            //声明要关注的队列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            System.out.println("Customer Waiting Received messages");
    
            //DefaultConsumer类实现了Consumer接口，通过传入一个频道，
            // 告诉服务器我们需要那个频道的消息，如果频道中有消息，就会执行回调函数handleDelivery
            Consumer consumer = new DefaultConsumer(channel) {
                public void handleDelivery(String consumerTag, Envelope envelope,
                                           AMQP.BasicProperties properties, byte[] body)
                        throws IOException {
                    String message = new String(body, "UTF-8");
                    System.out.println("Customer Received '" + message + "'");
                }
            };
            //自动回复队列应答 -- RabbitMQ中的消息确认机制
            channel.basicConsume(QUEUE_NAME, true, consumer);
        }
    }

![复制代码][copycode.gif]

执行

Producer.java

Producer Send +'Hello RabbitMQ'
    Producer Send +'Hello RabbitMQ'

Consumer.java

<table> 
 <tbody> 
  <tr> 
   <td> 
    <div>
      1 
    </div> 
    <div>
      2 
    </div></td> 
   <td> 
    <div> 
     <div> 
      <code>Customer Received </code> 
      <code>'Hello RabbitMQ'</code> 
     </div> 
     <div> 
      <code>Customer Received </code> 
      <code>'Hello RabbitMQ'</code> 
     </div> 
    </div></td> 
  </tr> 
 </tbody> 
</table>

[回到顶部][Link 1]

### 例子2 ###

首先写一个类，将产生产者和消费者统一为 EndPoint类型的队列。不管是生产者还是消费者，连接队列的代码都是一样的，这样可以通用一些。

EndPoint.java

![复制代码][copycode.gif]

//package co.syntx.examples.rabbitmq;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    /**
     * Represents a connection with a queue
     * @author syntx
     *
     */
    public abstract class EndPoint{
    
        protected Channel channel;
        protected Connection connection;
        protected String endPointName;
    
        public EndPoint(String endpointName) throws IOException{
            this.endPointName = endpointName;
    
            //Create a connection factory
            ConnectionFactory factory = new ConnectionFactory();
    
            //hostname of your rabbitmq server
            factory.setHost("100.51.15.10");
            factory.setUsername("admin");
            factory.setPassword("admin");
            factory.setPort(5672);
    
            //getting a connection
            try{
                connection = factory.newConnection();
            }catch (TimeoutException ex) {
                System.out.println(ex);
                connection = null;
            }
    
            //creating a channel
            channel = connection.createChannel();
    
            //declaring a queue for this channel. If queue does not exist,
            //it will be created on the server.
            channel.queueDeclare(endpointName, false, false, false, null);
        }
    
    
        /**
         * 关闭channel和connection。并非必须，因为隐含是自动调用的。
         * @throws IOException
         */
        public void close() throws IOException{
            try{
                this.channel.close();
            } catch (TimeoutException ex){
                System.out.println("ex" + ex);
            }
            this.connection.close();
        }
    }

![复制代码][copycode.gif]

Producer2.java

![复制代码][copycode.gif]

import java.io.IOException;
    import java.io.Serializable;
    
    import org.apache.commons.lang.SerializationUtils;
    
    public class Producer2 extends EndPoint{
    
        public Producer2(String endPointName) throws IOException{
            super(endPointName);
        }
    
        public void sendMessage(Serializable object) throws IOException {
            channel.basicPublish("",endPointName, null, SerializationUtils.serialize(object));
        }
    }

![复制代码][copycode.gif]

QueueConsumer.java

![复制代码][copycode.gif]

import java.io.IOException;
    import java.util.HashMap;
    import java.util.Map;
    
    import org.apache.commons.lang.SerializationUtils;
    
    import com.rabbitmq.client.AMQP.BasicProperties;
    import com.rabbitmq.client.Consumer;
    import com.rabbitmq.client.Envelope;
    import com.rabbitmq.client.ShutdownSignalException;
    
    
    public class QueueConsumer extends EndPoint implements Runnable, Consumer{
    
        public QueueConsumer(String endPointName) throws IOException{
            super(endPointName);
        }
    
        public void run() {
            try {
                //start consuming messages. Auto acknowledge messages.
                channel.basicConsume(endPointName, true,this);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        /**
         * Called when consumer is registered.
         */
        public void handleConsumeOk(String consumerTag) {
            System.out.println("Consumer "+consumerTag +" registered");
        }
    
        /**
         * Called when new message is available.
         */
        public void handleDelivery(String consumerTag, Envelope env,
                                   BasicProperties props, byte[] body) throws IOException {
            Map map = (HashMap)SerializationUtils.deserialize(body);
            System.out.println("Message Number "+ map.get("message number") + " received.");
    
        }
    
        public void handleCancel(String consumerTag) {}
        public void handleCancelOk(String consumerTag) {}
        public void handleRecoverOk(String consumerTag) {}
        public void handleShutdownSignal(String consumerTag, ShutdownSignalException arg1) {}
    }

![复制代码][copycode.gif]

Main.java

![复制代码][copycode.gif]

import java.io.IOException;
    import java.sql.SQLException;
    import java.util.HashMap;
    
    public class Main {
        public Main() throws Exception{
    
            QueueConsumer consumer = new QueueConsumer("queue");
            Thread consumerThread = new Thread(consumer);
            consumerThread.start();
    
            Producer2 producer = new Producer2("queue");
    
            for (int i = 0; i < 5; i++) {
                HashMap message = new HashMap();
                message.put("message number", i);
                producer.sendMessage(message);
                System.out.println("Message Number "+ i +" sent.");
            }
        }
    
        public static void main(String[] args) throws Exception{
            new Main();
            System.out.println("##############end...");
        }
    }

![复制代码][copycode.gif]

分类: [java][]

[1]: https://www.cnblogs.com/kaituorensheng/p/6591581.html#_label0
[2]: https://www.cnblogs.com/kaituorensheng/p/6591581.html#_label1
[Link 1]: https://www.cnblogs.com/kaituorensheng/p/6591581.html#_labelTop
[copycode.gif]: /images/20220525/a6bf8689977f43da9e1e5f5844142d60.png
[java]: http://www.cnblogs.com/kaituorensheng/category/586157.html