rabbitmq 发送端和消费端实例代码

wade6 · wade6 · commit 96fd52a7ac3e · 2019-10-06T11:09:01.000+08:00
diff --git a/pom.xml b/pom.xml
@@ -38,6 +38,14 @@
 
 	<dependencies>
 
+		<!-- rabbit mq -->
+		<dependency>
+			<groupId>com.rabbitmq</groupId>
+			<artifactId>amqp-client</artifactId>
+			<version>5.1.2</version>
+		</dependency>
+		<!-- rabbit mq end -->
+
 		<dependency>
 			<groupId>org.jsoup</groupId>
 			<artifactId>jsoup</artifactId>
diff --git a/src/main/java/com/code/repository/study/http/HscodeGatherer.java b/src/main/java/com/code/repository/study/http/HscodeGatherer.java
@@ -38,8 +38,8 @@ public class HscodeGatherer implements Callable<List<Map<String,String>>> {
         this.hscode4 = hscode4;
     }
 
-    private static int corePoolSize = 20;  // 主流线程个数
-    private static int maximumPoolSize = 50; // 线程最大个数
+    private static int corePoolSize = 50;  // 主流线程个数
+    private static int maximumPoolSize = 100; // 线程最大个数
     private static long keepAliveTime = 1000L; // 大于主流线程个数的线程空闲的过期时间  wait for new tasks before terminating
     private static TimeUnit unit = TimeUnit.MILLISECONDS; // 时间单元
     private static BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<Runnable>(); // 工作队列，有三种类SynchronousQueue、LinkedBlockingQueue(在所有 corePoolSize 线程都忙时新任务在队列中等待,maximumPoolSiz失效)、ArrayBlockingQueue，分别对应同步队列、无界队列、有界队列。
diff --git a/src/main/java/com/code/repository/study/rabbitmq/ReciverMQ.java b/src/main/java/com/code/repository/study/rabbitmq/ReciverMQ.java
@@ -0,0 +1,36 @@
+package com.code.repository.study.rabbitmq;
+
+import com.rabbitmq.client.*;
+
+import java.io.IOException;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * rabbit MQ 接收端
+ */
+public class ReciverMQ {
+
+    private final static String QUEUE_NAME = "Hello";
+
+    public static void main(String[] args) throws IOException, TimeoutException {
+
+        ConnectionFactory factory = new ConnectionFactory();
+        factory.setHost("localhost");
+        factory.setPort(5672);
+        factory.setUsername("guest");
+        factory.setPassword("guest");
+        Connection connection = factory.newConnection();
+        Channel channel = connection.createChannel();
+        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
+        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
+        Consumer consumer = new DefaultConsumer(channel) {
+            @Override
+            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
+                String message = new String(body, "UTF-8");
+                System.out.println(" [x] Received '" + message + "'");
+            }
+        };
+        channel.basicConsume(QUEUE_NAME, true, consumer);
+
+    }
+}
diff --git a/src/main/java/com/code/repository/study/rabbitmq/SenderMQ.java b/src/main/java/com/code/repository/study/rabbitmq/SenderMQ.java
@@ -0,0 +1,46 @@
+package com.code.repository.study.rabbitmq;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+
+import java.io.IOException;
+import java.util.Scanner;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * rabbit MQ 发送端
+ */
+public class SenderMQ {
+
+    private final static String QUEUE_NAME = "Hello";
+
+    public static void main(String[] args) throws IOException, TimeoutException {
+        // connection是socket连接的抽象，并且为我们管理协议版本协商（protocol version negotiation），
+        // 认证（authentication ）等等事情。这里我们要连接的消息代理在本地，因此我们将host设为“localhost”。
+        // 如果我们想连接其他机器上的代理，只需要将这里改为特定的主机名或IP地址。
+        ConnectionFactory factory = new ConnectionFactory();
+        factory.setHost("localhost");
+        factory.setPort(5672); //默认端口号
+        factory.setUsername("guest");//默认用户名
+        factory.setPassword("guest");//默认密码
+        Connection connection = factory.newConnection();
+        Channel channel = connection.createChannel();
+        // 接下来，我们创建一个channel，绝大部分API方法需要通过调用它来完成。
+        // 发送之前，我们必须声明消息要发往哪个队列，然后我们可以向队列发一条消息：
+        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
+
+        Scanner input=new Scanner(System.in);
+
+        while(input.hasNextLine()){
+            String message = input.nextLine();
+            channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
+            System.out.println(" [x] Sent '" + message + "'");
+        }
+
+        System.out.println("send is end!");
+
+        channel.close();
+        connection.close();
+    }
+}
