rabbitmq demo

rmqdemo.kt

import com.rabbitmq.client.AMQP
import com.rabbitmq.client.ConnectionFactory
import com.rabbitmq.client.DefaultConsumer
import com.rabbitmq.client.Envelope
import java.nio.charset.Charset
import java.util.*
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit

fun main(args: Array<String>) {
    val exchange = "jhuan.follow"

    val cf = ConnectionFactory()
    cf.host = "218.168.168.239"

    val conn = cf.newConnection()
    val channel = conn.createChannel()

    channel.exchangeDeclare(exchange, "topic")

    val queue1 = channel.queueDeclare().queue
    val queue2 = channel.queueDeclare().queue

    println("got queue1 $queue1, queue2 $queue2")

    val executor = Executors.newSingleThreadScheduledExecutor()

    executor.scheduleAtFixedRate({
        val rand = Random().nextInt(10)
        if (rand > 5) {
            channel.basicPublish(exchange, "position.onupdate.sltf", null, "hi update".toByteArray())
        } else {
            channel.basicPublish(exchange, "position.oncreate", null, "hi create".toByteArray())
        }
    }, 0, 1000, TimeUnit.MILLISECONDS)


    channel.queueBind(queue1, exchange, "position.onupdate.*")
    channel.queueBind(queue2, exchange, "position.oncreate")

    channel.basicConsume(queue1, true, object : DefaultConsumer(channel) {
        override fun handleDelivery(consumerTag: String?, envelope: Envelope?, properties: AMQP.BasicProperties?, body: ByteArray?) {
            val msg = String(body!!, Charset.defaultCharset())
            println("got onupdate $msg")
        }
    })

    channel.basicConsume(queue2, true, object : DefaultConsumer(channel) {
        override fun handleDelivery(consumerTag: String?, envelope: Envelope?, properties: AMQP.BasicProperties?, body: ByteArray?) {
            val msg = String(body!!, Charset.defaultCharset())
            println("got oncreate $msg")
        }
    })


}

output:

got queue1 amq.gen-OTUkxLSIRUuQ-gT_hbgqjg, queue2 amq.gen-ynP2x1XvqfHlUEP_BMM6kw
got onupdate hi update
got oncreate hi create
got oncreate hi create
got oncreate hi create
got oncreate hi create
got oncreate hi create
got oncreate hi create
got onupdate hi update
got oncreate hi create
got oncreate hi create
got onupdate hi update
got oncreate hi create
got onupdate hi update
got onupdate hi update
got oncreate hi create