[Spring Boot] Spring과 Android에서 RabbitMQ 사용해보기

목차

1. Intro
2. RabbitMQ 설치
3. Spring Boot Publisher 만들기
4. Android Subscriber 만들기
5. 동작 확인

Intro

프로젝트 진행 중 실시간으로 주문 상태를 알려주는 서비스를 제공하는 요구사항이 추가됐다.

어떤 기술을 사용해야 할 지 찾아보다가, 메시지 큐를 사용하면 된다는 것을 알게 되었다.

Kafka, RabbitMQ, Redis, Mosquitto 등 다양한 기술 중 RabbitMQ가 지금 프로젝트에 가장 적합하다고 판단되어 적용하기로 했다.

시나리오 상 Spring이 Publisher, Android가 Subscriber이기 때문에 이 예제도 동일하게 진행했다.

 

소스코드는 Github에 업로드 되어있다.

GitHub - DevJaewoo/blog-code

---

RabbitMQ 설치

Docker를 사용해서 간단하게 설치할 수 있다.

management가 없으면 관리 툴이 제공되지 않는다. 만약 필요없으면 -management 부분을 제거하면 된다.

docker pull rabbitmq:3-management

Docker 설치 및 사용법은 아래의 글에서 확인할 수 있다.

[Docker] 도커 간단 사용법

 

이미지 다운로드가 완료됐으면 컨테이너를 실행시키자. 5672번 포트는 RabbitMQ, 15672번 포트는 관리 페이지다.

-h 을 넣은 이유는 데이터 추적을 위해 hostname 옵션을 줘야 하기 때문이다.

What this means for usage in Docker is that we should specify -h / --hostname explicitly for each daemon so that we don't get a random hostname and can keep track of our data:

docker run --name rabbitmq_server -it -d -h rabbitmq-server -p 5672:5672 -p 15672:15672 rabbitmq:3-management

 

이러면 설치는 끝이다.

관리페이지가 잘 뜨는지 확인만 하고 다음으로 넘어가자.

로그인 페이지가 보이면 된다. 로그인은 guest / guest로 할 수 있다.

---

Spring Boot Publisher 만들기

Dependency는 Spring Web, Lombok, Spring for RabbitMQ 3개를 추가하면 된다.

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-amqp'
    implementation 'org.springframework.boot:spring-boot-starter-web'
    compileOnly 'org.projectlombok:lombok'
    annotationProcessor 'org.projectlombok:lombok'
    testImplementation 'org.springframework.boot:spring-boot-starter-test'
    testImplementation 'org.springframework.amqp:spring-rabbit-test'
}

 

application.yml

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest

 

SpringConfig.java

@Configuration
public class SpringConfig {

    @Value("${spring.rabbitmq.host:localhost}")
    private String host;

    @Value("${spring.rabbitmq.port:5672}")
    private int port;

    @Value("${spring.rabbitmq.username}")
    private String username;

    @Value("${spring.rabbitmq.password}")
    private String password;

    @Bean
    public TopicExchange topicExchange() {
        String EXCHANGE_NAME = "com.exchange";
        return new TopicExchange(EXCHANGE_NAME);
    }

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host, port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);

        return connectionFactory;
    }

    @Bean
    public AmqpAdmin amqpAdmin() {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory());
        rabbitAdmin.declareExchange(topicExchange());
        return rabbitAdmin;
    }
}

Exchange는 하나만 두고 여러 Queue에 route해주도록 구성할 것이기 때문에 TopicExchange는 Bean으로 등록했다.

RabbitAdmin이 있어야 Exchange, Queue, Binding을 생성할 수 있기 때문에 이미 만들어둔 것을 사용할게 아니라면 AmqpAdmin도 필요하다.

RabbitTemplate로는 기존의 Exchange, Queue만 사용할 수 있고 생성할 순 없다.

 

TestController.java

@Slf4j
@RestController
@RequiredArgsConstructor
public class TestController {

    private final TopicExchange topicExchange;
    private final RabbitTemplate rabbitTemplate;
    private final AmqpAdmin rabbitAdmin;

    @PostMapping("/rabbit/register")
    public String register(@RequestBody QueueRequest queueRequest) {
        String queueName = queueRequest.getQueue();
        String routingKey = "com.devjaewoo.order.*";
        log.info("Binding queue " + queueName + " with Routing key " + routingKey);

        Binding binding = BindingBuilder.bind(new Queue(queueName)).to(topicExchange).with(routingKey);
        rabbitAdmin.declareBinding(binding);

        return "{\"result\": \"Success\"}";
    }

    @GetMapping("/rabbit/publish/{id}")
    public String publish(@PathVariable Long id) {
        String message = "Ordered ID: " + id;
        String routingKey = "com.devjaewoo.order." + id;

        log.info("Publish message " + message + " to " + routingKey);
        rabbitTemplate.convertAndSend("com.exchange", routingKey, message);

        return "Publish Success";
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    static class QueueRequest {
        private String queue;
    }
}

Android에서 앱 시작 시 자신의 UUID로 Queue를 생성하고, /rabbit/register URL에 UUID와 함께 POST 요청을 보내면

Spring에서 Exchange와 Queue를 Binding 한다.

그 다음 /rabbit/publish/{id}로 GET 요청이 들어오면 ID값을 Android에게 Publish 하도록 설정했다.

 

코드를 다 작성했으면 실행하고 관리자 페이지에 들어가 exchange가 추가됐는지 확인해보자.

리스트 맨 마지막에 com.exchange가 추가됐다.

---

Android Subscriber 만들기

Android에선 간단하게 앱이 실행되면 RabbitMQ에 랜덤 UUID로 Queue를 생성하고, 서버에 Post로 보내줄 것이다.

그리고 RabbitMQ에 의해 Callback 함수가 실행되면, 로그로 출력해보도록 하겠다.

 

Dependency

dependencies{

    implementation 'androidx.core:core-ktx:1.7.0'
    implementation 'androidx.appcompat:appcompat:1.5.0'
    implementation 'com.google.android.material:material:1.6.1'
    implementation 'androidx.constraintlayout:constraintlayout:2.1.4'
    implementation 'com.rabbitmq:amqp-client:5.15.0'
    testImplementation 'junit:junit:4.13.2'
    androidTestImplementation 'androidx.test.ext:junit:1.1.3'
    androidTestImplementation 'androidx.test.espresso:espresso-core:3.4.0'

    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-android:1.6.1'
    implementation 'com.android.volley:volley:1.2.1'
}

네트워크 작업을 위한 Coroutine과 Volley가 추가됐다.

 

ApplicationManager.kt

class ApplicationManager : Application() {

    companion object {
        lateinit var applicationContext: Context
    }

    override fun onCreate() {
        super.onCreate()
        ApplicationManager.applicationContext = applicationContext
    }
}

Volley에서 applicationContext를 사용하기 위해 추가해줬다.

 

RequestHandler.kt

const val TAG = "RequestHandler"
const val SERVER_URL = "http://192.168.25.7:8080"

object RequestHandler {
    var accessToken: String = ""
    private val requestQueue = Volley.newRequestQueue(ApplicationManager.applicationContext)

    private val defaultErrorListener = Response.ErrorListener { error ->
        if (error.networkResponse != null) {
            val body = JSONObject(String(error.networkResponse.data))
            val errorMessage = body.getString("message")
            Log.e(TAG, "defaultErrorListener: code: ${error.networkResponse.statusCode} message: $errorMessage")
            Toast.makeText(ApplicationManager.applicationContext, errorMessage, Toast.LENGTH_SHORT).show()
        } else {
            Log.e(TAG, "defaultErrorListener: null")
            Toast.makeText(ApplicationManager.applicationContext, "null", Toast.LENGTH_SHORT).show()
        }
    }

    fun request(
        url: String,
        jsonObject: JSONObject?,
        responseListener: Response.Listener<JSONObject>,
        errorListener: Response.ErrorListener?,
        requestWithToken: Boolean,
        method: Int
    ) {

        val requestURL = SERVER_URL + url
        Log.d(TAG, "request: $requestURL with data $jsonObject")

        val jsonObjectRequest = object : JsonObjectRequest(
            method,
            requestURL,
            jsonObject,
            responseListener,
            errorListener ?: defaultErrorListener
        ) {

            override fun getHeaders(): MutableMap<String, String> {
                return if (requestWithToken) HashMap<String, String>().apply { put("Authorization", "Bearer $accessToken") } else super.getHeaders()
            }
        }

        requestQueue.add(jsonObjectRequest)
    }
}

Volley Request를 좀 더 쉽게 하기 위한 Object다. 내가 예전에 만들어 둔걸 그대로 가져왔다.

 

MainActivity.kt

class MainActivity : AppCompatActivity() {
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)

        //네트워크 작업은 Main Thread에서 하면 Exception 터짐
        CoroutineScope(Dispatchers.IO).launch {
            kotlin.runCatching {
                try {
                    //랜덤 UUID 생성
                    val queueName = UUID.randomUUID().toString()

                    //RabbitMQ 서버와 연결
                    ConnectionFactory().apply { host = "192.168.25.7" }
                        .newConnection()
                        .createChannel().apply {
                            //UUID로 Queue 생성
                            queueDeclare(queueName, false, false, true, null)
                            
                            //생성한 Queue에 Callback Listener 등록
                            basicConsume(queueName, true,
                                { consumerTag, message ->
                                    Log.d(TAG, "DeliverCallback: tag: $consumerTag, message: ${message.body.toString(Charsets.UTF_8)}")
                                },
                                { consumerTag ->
                                    Log.d(TAG, "CancelCallback: $consumerTag")
                                }
                            )

                            //Spring 서버에 Queue UUID 전송
                            RequestHandler.request(
                                "/rabbit/register",
                                JSONObject().apply { put("queue", queueName) },
                                { response ->
                                    Log.d(TAG, "onCreate: $response")
                                },
                                null,
                                false,
                                Request.Method.POST
                            )
                        }

                } catch (e: Exception) {
                    Log.e(TAG, "onCreate: ${e.message}", e.cause)
                    e.printStackTrace()
                }
            }
        }
    }
}

코드가 좀 길지만 뜯어보면 하는게 별로 없다.

Coroutine에서 랜덤 UUID로 Queue를 생성하고, Callback 등록하고, 서버에 보내주는게 끝이다.

 

AndroidManifest.xml

<?xml version="1.0" encoding="utf-8"?>
<manifest xmlns:android="http://schemas.android.com/apk/res/android"
    xmlns:tools="http://schemas.android.com/tools"
    package="com.devjaewoo.androidabbitmqtest">

    <uses-permission android:name="android.permission.INTERNET"/>

    <application
        android:name=".ApplicationManager"
        android:allowBackup="true"
        android:dataExtractionRules="@xml/data_extraction_rules"
        android:fullBackupContent="@xml/backup_rules"
        android:icon="@mipmap/ic_launcher"
        android:label="@string/app_name"
        android:roundIcon="@mipmap/ic_launcher_round"
        android:supportsRtl="true"
        android:theme="@style/Theme.AndroidRabbitmqTest"
        android:usesCleartextTraffic="true"
        tools:targetApi="31">
        <activity
            android:name=".MainActivity"
            android:exported="true">
            <intent-filter>
                <action android:name="android.intent.action.MAIN" />

                <category android:name="android.intent.category.LAUNCHER" />
            </intent-filter>
        </activity>
    </application>

</manifest>

ApplicationManager를 Application으로 등록하고 INTERNET 권한을 추가했다.

HTTP Request를 보내기 위해 usesCleartextTraffic도 설정해줬다.

 

이제 서버가 켜진 상태로 Android 앱을 실행했을 때 Logcat에 {"result":"Success"}라는 응답이 도착하는지 확인해보자.

---

동작 확인

만약 Android 앱이 정상적으로 실행됐다면 RabbitMQ 관리자 페이지에 새로운 Queue가 생성됐을 것이다.

만약 안생겼다면 Windows 방화벽에서 포트가 열려있는지 확인해보자.

Queue가 생겼다면 http://localhost:8080/rabbit/publish/1234로 접속해서 Android에 Publish하고, Callback에서 로그가 정상적으로 출력되는지 확인해보자.

마지막 줄을 보면 ID: 1234로 잘 수신되는 것을 볼 수 있다.

---

참고자료

• https://hub.docker.com/_/rabbitmq