RMQ on Java

/**
 * Class untuk definisi function RMQ
 */
public class RMQ {

    // Load Global Configuration
    Settings settings = new Settings();

    // Setup RMQ Configuration (Ganti dengan setting akun RMQ nya)
    String user = "user"
    String pass = "pass"
    String host = "host"
    String vhost = "vhost"
    String exchanges_name = "exchange"
    String queue_name_publish = "queue_publish"
    String queue_name_subscribe = "queue_subscribe"
    String routingKey = "routing key"

    // Define new Connection Factory
    ConnectionFactory factory = new ConnectionFactory();

    /**
     * Setup Connection Factory
     * Load Setting Connection RMQ Sebagai Parameter Koneksi
     */
    public void setupConnectionFactory() {
        try {
            factory.setAutomaticRecoveryEnabled(false);
            factory.setUri("amqp://"+user+":"+pass+"@"+host);
            factory.setVirtualHost(vhost);
        } catch (KeyManagementException | NoSuchAlgorithmException | URISyntaxException e1) {
            e1.printStackTrace();
        }
    }

    /**
     * Publish data lewat RMQ
     * @param message
     */
    public void publish(String message) {
        try {
            StrictMode.ThreadPolicy policy = new StrictMode.ThreadPolicy.Builder().permitAll().build();
            StrictMode.setThreadPolicy(policy);

            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();

            String messageOn = message ;
            channel.basicPublish("", queue_name_publish,null,messageOn.getBytes());

        } catch (IOException e) {
            Log.d("Publish Error", e.getMessage());
        } catch (TimeoutException e) {
            Log.d("Publish Error", e.getMessage());
        } catch (Exception e) {
            Log.d("Publish Error", e.getMessage());
        }

    }

    /**
     * Optional, Send Speed for threading speed
     * @throws InterruptedException
     */
    public void SendSpeed() throws InterruptedException {
        Thread.sleep(500); //0.5 sec
    }

    /**
     * Fungsi untuk subscribe data RMQ
     * @param handler
     * @param subscribeThread
     */
    public void subscribe(final Handler handler, Thread subscribeThread)
    {
        subscribeThread = new Thread(new Runnable() {
            @Override
            public void run() {
                while(true) {
                    try {
                        Connection connection = factory.newConnection();
                        Channel channel = connection.createChannel();
                        channel.basicQos(0);
                        channel.queueBind(queue_name_subscribe, exchanges_name, routingKey);
                        QueueingConsumer consumer = new QueueingConsumer(channel);
                        channel.basicConsume(queue_name_subscribe, true, consumer);
                        QueueingConsumer.Delivery delivery = consumer.nextDelivery();

                        if (delivery != null){

                            try{

                                String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                                Log.d("ConsumeDataRMQ", "MessageConsumed" + message);

                                Message msg = handler.obtainMessage();
                                Bundle bundle = new Bundle();

                                bundle.putString("msg", message);
                                msg.setData(bundle);
                                handler.sendMessage(msg);

                                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

                            }catch (Exception e){
                                channel.basicReject(delivery.getEnvelope().getDeliveryTag(),true);
                            }
                        }
                    } catch (InterruptedException e) {
                        break;
                    } catch (Exception e1) {
                        Log.d("", "Connection broken: " + e1.getClass().getName());
                        try {
                            Thread.sleep(4000); //sleep and then try again
                        } catch (InterruptedException e) {
                            break;
                        }
                    }
                }
            }
        });
        subscribeThread.start();
    }
}

--------------
Cara Make di Main Activity
1. Buat fungsi show notification dulu
    public void showNotification( String title, String body) {
        Context context = getApplicationContext();
        Intent intent = getIntent();

        NotificationManager notificationManager = (NotificationManager) context.getSystemService(Context.NOTIFICATION_SERVICE);
        int notificationId = 1;
        String channelId = "channel-01";
        String channelName = "Channel Name";
        int importance = NotificationManager.IMPORTANCE_HIGH;

        if (android.os.Build.VERSION.SDK_INT >= android.os.Build.VERSION_CODES.O) {
            NotificationChannel mChannel = new NotificationChannel(
                    channelId, channelName, importance);
            notificationManager.createNotificationChannel(mChannel);
        }

        NotificationCompat.Builder mBuilder = new NotificationCompat.Builder(context, channelId)
                .setSmallIcon(R.drawable.bawasluicon)
                .setContentTitle(title)
                .setContentText(body);

        TaskStackBuilder stackBuilder = TaskStackBuilder.create(context);
        stackBuilder.addNextIntent(intent);
        PendingIntent resultPendingIntent = stackBuilder.getPendingIntent(
                0,
                PendingIntent.FLAG_UPDATE_CURRENT
        );
        mBuilder.setContentIntent(resultPendingIntent);

        notificationManager.notify(notificationId, mBuilder.build());
    }

2. Subscribe ke RMQ ketika menerima data bakal panggil Subscribe notification di activity
2.1 Inisialisasi dulu di activity dengan
   RMQ rmq = new RMQ();

2.2 Kemudian di On Create Panggil Inisialisasi Koneksi RMQ dan panggil fungsi subscribe notification
   rmq.setupConnectionFactory();
   subscribeNotification();

2.2.1 Fungsi untuk subscribe notificationnya seperti ini
    private void subscribeNotification() {
        final Handler incomingMessageHandler = new Handler() {
            @Override
            public void handleMessage(Message msg) {
                String title = "Pengumuman";
                String message = msg.getData().getString("msg");
                Log.d("RMQMessage", message);

                showNotification(title, message);
            }
        };
        rmq.subscribe(incomingMessageHandler,subscribeThread);
    }