Week14

week14/rabbit-mq/hello_rabbitmq_java/.gitignore

@@ -0,0 +1,38 @@
+target/
+!.mvn/wrapper/maven-wrapper.jar
+!**/src/main/**/target/
+!**/src/test/**/target/
+
+### IntelliJ IDEA ###
+.idea/modules.xml
+.idea/jarRepositories.xml
+.idea/compiler.xml
+.idea/libraries/
+*.iws
+*.iml
+*.ipr
+
+### Eclipse ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+.sts4-cache
+
+### NetBeans ###
+/nbproject/private/
+/nbbuild/
+/dist/
+/nbdist/
+/.nb-gradle/
+build/
+!**/src/main/**/build/
+!**/src/test/**/build/
+
+### VS Code ###
+.vscode/
+
+### Mac OS ###
+.DS_Store

week14/rabbit-mq/hello_rabbitmq_java/pom.xml

@@ -0,0 +1,57 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
+  <groupId>dev.mq</groupId>
+  <artifactId>hello_rabbitmq_java</artifactId>
+  <version>1.0-SNAPSHOT</version>
+  <packaging>jar</packaging>
+
+  <name>hello_rabbitmq_java</name>
+  <url>http://maven.apache.org</url>
+
+  <properties>
+    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+  </properties>
+
+  <dependencies>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <version>3.8.1</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.rabbitmq</groupId>
+      <artifactId>amqp-client</artifactId>
+      <version>5.18.0</version>
+    </dependency>
+
+    <!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-simple -->
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-simple</artifactId>
+      <version>2.0.16</version>
+      <scope>test</scope>
+    </dependency>
+
+    <!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-api -->
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-api</artifactId>
+      <version>2.0.16</version>
+    </dependency>
+  </dependencies>
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-compiler-plugin</artifactId>
+        <configuration>
+          <source>8</source>
+          <target>8</target>
+        </configuration>
+      </plugin>
+    </plugins>
+  </build>
+</project>

week14/rabbit-mq/hello_rabbitmq_java/src/main/java/dev/mq/App.java

@@ -0,0 +1,13 @@
+package dev.mq;
+
+/**
+ * Hello world!
+ *
+ */
+public class App 
+{
+    public static void main( String[] args )
+    {
+        System.out.println( "Hello World!" );
+    }
+}

week14/rabbit-mq/hello_rabbitmq_java/src/main/java/dev/mq/steo01/Receiver.java

@@ -0,0 +1,40 @@
+package dev.mq.steo01;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.TimeoutException;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.DeliverCallback;
+
+/**
+ * Receiver - 메시지 소비자 (Consumer)행
+ *
+ */
+public class Receiver {
+	private static final String QUEUE_NAME = "hello";
+
+	public static void main(String[] args) throws IOException, TimeoutException {
+
+		ConnectionFactory factory = new ConnectionFactory();
+		factory.setHost("localhost");
+
+		// RabbitMQ 서버(5672)와 커넥션 연결 수행
+		Connection connection = factory.newConnection();
+		// 채녈 생성
+		Channel channel = connection.createChannel();
+
+		// 큐 생성
+		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
+		System.out.println(" [Receiver] Waiting ... ");
+
+		DeliverCallback deliverCallback = (consumerTag, delivery) -> {
+			String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
+			System.out.println("[Receiver] 메시지를 받음: " + message);
+		};
+
+		channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {});
+	}
+}

week14/rabbit-mq/hello_rabbitmq_java/src/main/java/dev/mq/steo01/Sender.java

@@ -0,0 +1,39 @@
+package dev.mq.steo01;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.TimeoutException;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+
+/**
+ * Sender - 메시지 발행자 (Publisher)
+ *
+ * 퍼블리셔는 RabbitMq와 연결, 메시지를 전송하는 처리 수행
+ * 컨슈머는 메시지 큐로부터 퍼블리셔가 적재한 메시지를 소비하는 처리 수행
+ */
+public class Sender {
+	private static final String QUEUE_NAME = "hello";
+
+	public static void main(String[] args) throws IOException, TimeoutException {
+		ConnectionFactory factory = new ConnectionFactory();
+		factory.setHost("localhost");
+
+		// RabbitMQ 서버(5672)와 커넥션 연결 수행
+		Connection connection = factory.newConnection();
+		// 채녈 생성
+		Channel channel = connection.createChannel();
+
+		// 큐 생성
+		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
+
+		String messgae = "Hell World!";
+
+		// Publish
+		channel
+			.basicPublish("", QUEUE_NAME, null, messgae.getBytes(StandardCharsets.UTF_8));
+		System.out.println(" [Publisher] Sent = " + messgae);
+	}
+}

week14/rabbit-mq/hello_rabbitmq_java/src/main/java/dev/mq/step02/NewTask.java

@@ -0,0 +1,46 @@
+package dev.mq.step02;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.MessageProperties;
+
+import java.io.IOException;
+import java.util.concurrent.TimeoutException;
+
+/*
+    작업(Task)을 작업 큐(Work queue)에 스케줄링하는 프로그램
+    문자열값이 어떤 복잡한 작업이라고 가정할 때,
+    ex. Hello...이라고 하면
+    . 하나는 1초가 소요되는 작업이라고 가정
+
+    따라서 Hello...는 3초가 걸리는 작업이라고 가정
+
+    실행 인자값(args) 활용해서 Hello...과 같은 값 입력
+ */
+public class NewTask {
+
+	private static final String TASK_QUEUE_NAME = "task_queue";
+
+	public static void main(String[] args) throws IOException, TimeoutException {
+		// 서버 연결
+		ConnectionFactory factory = new ConnectionFactory();
+		factory.setHost("localhost");
+		try (Connection connection = factory.newConnection();
+			 Channel channel = connection.createChannel()) {
+
+			// 큐 생성
+			boolean durable = true;
+			channel.queueDeclare(TASK_QUEUE_NAME, durable, false, false, null);
+			// 실행 옵션으로 메시지를 생성, ex. Hello...
+			String message = String.join(" ", args);
+
+			// 메시지 적재
+			channel.basicPublish("", TASK_QUEUE_NAME,
+				MessageProperties.PERSISTENT_TEXT_PLAIN,
+				message.getBytes("UTF-8"));
+			System.out.println(" [Publisher] Sent " + message);
+		}
+
+	}
+}

week14/rabbit-mq/hello_rabbitmq_java/src/main/java/dev/mq/step02/Worker.java

@@ -0,0 +1,62 @@
+package dev.mq.step02;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.DeliverCallback;
+
+import java.io.IOException;
+import java.util.concurrent.TimeoutException;
+
+/*
+    작업 큐에서 꺼낸 작업의 메시지를 처리하는 역할
+ */
+public class Worker {
+
+	private static final String TASK_QUEUE_NAME = "task_queue";
+
+	public static void main(String[] args) throws IOException, TimeoutException {
+		ConnectionFactory factory = new ConnectionFactory();
+		factory.setHost("localhost");
+		final Connection connection = factory.newConnection();
+		final Channel channel = connection.createChannel();
+
+		boolean durable = true;
+		channel.queueDeclare(TASK_QUEUE_NAME, durable, false, false, null);
+		System.out.println("[Consumer(Worker)] 메시지를 기다리는 중..");
+
+		channel.basicQos(1);
+
+		DeliverCallback deliverCallback = (consumerTag, delivery) -> {
+			// 메시지 추출
+			String message = new String(delivery.getBody(), "UTF-8");
+
+			System.out.println(" [Consumer(Worker)] Received '" + message + "'");
+			try {
+				doWork(message); // 받은 메시지 처리 작업 수행
+			} finally {
+				System.out.println(" [Consumer(Worker)] Done");
+				channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
+			}
+		};
+
+		boolean autoAcknowledgement = false;
+		channel.basicConsume(TASK_QUEUE_NAME, autoAcknowledgement, deliverCallback, consumerTag -> {
+		});
+	}
+
+	// Task 처리 메서드, Hello...일 경우 .하나당 1초씩 지연해서 처리해야함
+	private static void doWork(String task) {
+		int cnt = 0;
+		for (char ch : task.toCharArray()) {
+			if (ch == '.') { // .(dot) 한 개당 1초씩 지연
+				try {
+					System.out.println(++cnt + "초 동안 작업 수행중");
+					Thread.sleep(1000);
+				} catch (InterruptedException _ignored) {
+					Thread.currentThread().interrupt();
+				}
+			}
+		}
+	}
+}

week14/rabbit-mq/hello_rabbitmq_java/src/main/java/dev/mq/step03/EmitLogs.java

@@ -0,0 +1,45 @@
+package dev.mq.step03;
+
+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.MessageProperties;
+
+/*
+    작업(Task)을 작업 큐(Work queue)에 스케줄링하는 프로그램
+    문자열값이 어떤 복잡한 작업이라고 가정할 때,
+    ex. Hello...이라고 하면
+    . 하나는 1초가 소요되는 작업이라고 가정
+
+    따라서 Hello...는 3초가 걸리는 작업이라고 가정
+
+    실행 인자값(args) 활용해서 Hello...과 같은 값 입력
+ */
+public class EmitLogs {
+
+
+	private static final String EXCHANGE_NAME = "logs";
+
+	public static void main(String[] args) throws IOException, TimeoutException {
+		// 서버 연결
+		ConnectionFactory factory = new ConnectionFactory();
+		factory.setHost("localhost");
+		try (Connection connection = factory.newConnection();
+			 Channel channel = connection.createChannel()) {
+
+			String message = args.length < 1 ? "info: Hello World!" :
+				String.join(" ", args);
+
+
+			// Exchange 생성
+			channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
+
+			channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
+			System.out.println(" [x] Sent '" + message + "'");
+		}
+
+	}
+}

week14/rabbit-mq/hello_rabbitmq_java/src/main/java/dev/mq/step03/ReceiveLogs.java

@@ -0,0 +1,34 @@
+package dev.mq.step03;
+
+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.DeliverCallback;
+import com.rabbitmq.client.MessageProperties;
+
+public class ReceiveLogs {
+
+	private static final String EXCHANGE_NAME = "logs";
+
+	public static void main(String[] argv) throws Exception {
+		ConnectionFactory factory = new ConnectionFactory();
+		factory.setHost("localhost");
+		Connection connection = factory.newConnection();
+		Channel channel = connection.createChannel();
+
+		channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
+		String queueName = channel.queueDeclare().getQueue();
+		channel.queueBind(queueName, EXCHANGE_NAME, "");
+
+		System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
+
+		DeliverCallback deliverCallback = (consumerTag, delivery) -> {
+			String message = new String(delivery.getBody(), "UTF-8");
+			System.out.println(" [x] Received '" + message + "'");
+		};
+		channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });
+	}
+}