Отписаться от постоянных подписчиков с ActiveMQ

Question

Отписаться от постоянных подписчиков с ActiveMQ

Я пытаюсь НЕ ПОДПИСАТЬ постоянных подписчиков на ТЕМЫ.

Мое приложение - это своего рода социальная сеть: каждый пользователь - это тема для других пользователей. Таким образом, каждый раз, когда пользователь что-то делает, его друзья получают уведомление. Конечно, подписчик может отписаться от темы, желая больше получать уведомления о пользователе.

Каждый раз, когда я пытаюсь отписать подписчика от темы, я получаю сообщение об ошибке, в котором говорится: "javax.jms.JMSException: надежный потребитель используется"

Вот мои 2 класса, один отправитель и один приемник. Может кто-нибудь сказать мне, что я делаю не так??

SENDER Class:

package com.citizenweb.classes;

import java.util.Date;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageFormatException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.ObjectMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQSession;

import com.citizenweb.interfaces.PostIF;
import com.citizenweb.interfaces.UserIF;

public class Sender {

    private ActiveMQConnectionFactory factory = null;
    private ActiveMQConnection connection = null;
    private ActiveMQSession session = null;
    private Destination destination = null;
    private MessageProducer producer = null;

    public Sender() {
    }

    public void connect(){
        try{
            factory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_BROKER_URL);
            // TODO Mécanisme de sécurité d'ActiveMQ à rétablir en production
            factory.setTrustAllPackages(true);
            connection = (ActiveMQConnection) factory.createConnection();
            connection.start();
            session = (ActiveMQSession) connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        } catch (JMSException e){
            e.printStackTrace();
        }
    }

    public void sendPost(UserIF user,PostIF post) {
        if(session==null){connect();}
        try {
            destination = session.createTopic(user.toString());
            producer = session.createProducer(destination);
            ObjectMessage postMessage = session.createObjectMessage();
            postMessage.setObject(post);
            producer.send(postMessage);
            System.out.println("\n SENDER Object message sent");



        } catch (MessageFormatException e) {
            e.printStackTrace();
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    public void sendInformation(UserIF user,String info){
        if(session==null){connect();}
        try {
            destination = session.createTopic(user.toString());
            producer = session.createProducer(destination);
            TextMessage infoMessage = session.createTextMessage();
            infoMessage.setText(info);
            producer.send(infoMessage);
            System.out.println("\n SENDER Information message sent");
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    /**
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {

        UserIF u1, u2, u3;
        String[] nom = new String[5];
        String[] prenom = new String[5];
        String[] login = new String[5];
        String[] password = new String[5];
        Date[] naiss = new Date[5];
        String[] mail = new String[5];
        for (int i = 0; i < 5; i++) {
            nom[i] = "nom_" + i;
            prenom[i] = "prenom_" + i;
            login[i] = "login_" + i;
            password[i] = "password_" + i;
            naiss[i] = new Date();
            mail[i] = "mail_" + i;
        }

        System.out.println("\n SENDER AFFECTATION DES NOMS");
        u1 = new User(nom[0], prenom[0], login[0], password[0], naiss[0], mail[0]);
        u2 = new User(nom[1], prenom[1], login[1], password[1], naiss[1], mail[1]);
        u3 = new User(nom[2], prenom[2], login[2], password[2], naiss[2], mail[2]);


        Sender sender = new Sender();

        sender.sendInformation(u1, "U1 notification");
        sender.sendInformation(u2, "U2 notification");
        sender.sendInformation(u3, "U3 notification");
        //PostIF post = new Post("Mon Post","Ceci est mon message",u1,u1,"Classe Sender",((User) u1).getIdUser(),0);
        //sender.sendPost(user, post);
        sender.session.close();
        sender.connection.close();

    }

}

ПОЛУЧАТЕЛЬ Класс:

package com.citizenweb.classes;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQSession;
import org.apache.activemq.broker.region.Destination;
import com.citizenweb.interfaces.PostIF;
import com.citizenweb.interfaces.UserIF;
import com.citizenweb.classes.Post;

public class Receiver implements MessageListener, Serializable {

    private static final long serialVersionUID = 1L;
    private ActiveMQConnectionFactory factory = null;
    private ActiveMQConnection connection = null;
    private ActiveMQSession session = null;
    private Topic destination = null;
    private MessageConsumer consumer = null;

    UserIF userTopic = new User();
    UserIF userSubscriber = new User();
    List<Message> listeMsg = new ArrayList<Message>();

    public Receiver(UserIF subscriber) {
        this.userSubscriber = subscriber;
    }

    public void connect() {
        try {
            factory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_BROKER_URL);
            // TODO Mécanisme de sécurité d'ActiveMQ à rétablir en production
            factory.setTrustAllPackages(true);
            connection = (ActiveMQConnection) factory.createConnection();
            // ClientID :
            // https://qnalist.com/questions/2068823/create-durable-topic-subscriber
            connection.setClientID(userSubscriber.toString());
            connection.start();
            session = (ActiveMQSession) connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    public void receiveMessage(UserIF topic) {
        try {
            if (session == null) {
                connect();
            }
            destination = session.createTopic(topic.toString());
            String nomAbonnement = topic.toString() + "->" + userSubscriber.toString();
            //String nomAbonnement = userSubscriber.toString();
            consumer = session.createDurableSubscriber(destination, nomAbonnement);
            consumer.setMessageListener(this);
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    public void unsubscribe(UserIF topic) {
        try {
            if (session == null) {
                connect();
            }
            System.out.println("\n RECEIVER Désinscription du topic " + topic.toString());
            //consumer.close();
            String nomAbonnement = topic.toString() + "->" + userSubscriber.toString();
            //String nomAbonnement = userSubscriber.toString();
            System.out.println("\n RECEIVER Abonnement à clore = " + nomAbonnement);
            session.unsubscribe(nomAbonnement);
            System.out.println("\n RECEIVER " + userSubscriber.toString() + " s'est désinscrit de " + nomAbonnement);
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void onMessage(Message message) {
        System.out.println("\n RECEIVER OnMessage triggered for " + userSubscriber.toString());
        listeMsg.add(message);
        System.out.println("\n RECEIVER Nombre de messages reçus par " + userSubscriber + " = " + listeMsg.size());
        String classe = message.getClass().getSimpleName();
        System.out.println("\n RECEIVER Classe de message : " + classe);
        try {
            if (message instanceof TextMessage) {
                TextMessage text = (TextMessage) message;
                System.out.println("\n RECEIVER Information : " + text.getText());
            }
            if (message instanceof ObjectMessage) {
                System.out.println("\n RECEIVER ObjectMessage");
                ObjectMessage oMessage = (ObjectMessage) message;
                if (oMessage.getObject() instanceof PostIF) {
                    PostIF post = (PostIF) oMessage.getObject();
                    String s = ((Post) post).getCorpsMessage();
                    System.out.println("\n RECEIVER Post : " + s);
                }
            }
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws JMSException {

        /*
         * EACH USER IS A TOPIC FOR OTHER USERS
         * WHATEVER A USER DOES RESULTS IN A NOTIFICATION TO SUBSCRIBERS
        */

        //CREATE USER
        UserIF u1, u2, u3;
        String[] nom = new String[5];
        String[] prenom = new String[5];
        String[] login = new String[5];
        String[] password = new String[5];
        Date[] naiss = new Date[5];
        String[] mail = new String[5];
        for (int i = 0; i < 5; i++) {
            nom[i] = "nom_" + i;
            prenom[i] = "prenom_" + i;
            login[i] = "login_" + i;
            password[i] = "password_" + i;
            naiss[i] = new Date();
            mail[i] = "mail_" + i;
        }

        u1 = new User(nom[0], prenom[0], login[0], password[0], naiss[0], mail[0]);
        u2 = new User(nom[1], prenom[1], login[1], password[1], naiss[1], mail[1]);
        u3 = new User(nom[2], prenom[2], login[2], password[2], naiss[2], mail[2]);

        /*
         * MAKE EACH USER A SUBSCRIBER
         */
        Receiver receiver1 = new Receiver(u1);
        Receiver receiver2 = new Receiver(u2);
        Receiver receiver3 = new Receiver(u3);

        /*
         * PUT A MESSAGE LISTENER FOR EACH USER
         */
        receiver1.receiveMessage(u2);
        receiver1.receiveMessage(u3);
        receiver2.receiveMessage(u1);
        receiver2.receiveMessage(u3);
        receiver3.receiveMessage(u1);
        receiver3.receiveMessage(u2);

        /*
         * CALL THE SENDER CLASS TO SEND MESSAGES
         */
        try {
            Sender.main(args);
        } catch (Exception e1) {
            e1.printStackTrace();
        }

        /*
         * A SLEEP TO HAVE ENOUGH TIME TO LOOK AT THE ACTIVEMQ CONSOLE
         * CAN BE REMOVE
         */
        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
            return;
        }

        /*
         * UNSUBSCRIBE SUBSCRIBERS FROM TOPICS
         */
        receiver1.unsubscribe(u2);
        receiver1.unsubscribe(u3);
        receiver2.unsubscribe(u1);
        receiver2.unsubscribe(u3);
        receiver3.unsubscribe(u1);
        receiver3.unsubscribe(u2);
    }

}

0

java jms activemq-classic jms-topic durable-subscription

Источник

user4587314 16 сен '16 в 12:51

2 ответа

Другие вопросы по тегам java jms activemq-classic jms-topic durable-subscription

user4587314 19 сен '16 в 11:49 2016-09-19 11:49 · Answer 1 · 2016-09-19 11:49

Каждому соединению нужен уникальный ClientID: connection.setClientID ("clientID");

Моя ошибка состояла в том, чтобы неправильно понять эту уникальность для данного клиента.

Когда клиент подписывается на тему, существует одна связь для этой темы. Таким образом, для данного клиента, подписанного на 3 темы (например), требуется 3 ClientID, поскольку необходимо 3 соединения. ClientID должен быть уникальным, поскольку он идентифицирует одно соединение одного клиента для одной темы.

Вот почему у меня было так много JMSException говоря, что "Durable Consumer был в использовании", когда я хотел прекратить подписку.

user375722 16 сен '16 в 13:01 2016-09-16 13:01 · Answer 2 · 2016-09-16 13:01

Вы можете отменить подписку на длительную подписку только в том случае, если в настоящее время нет активного подписчика на нее. Похоже, ваш код создает несколько подписок и не останавливает потребителей, поэтому, конечно, отмена подписки не удастся, если вы закроете потребителей и затем откажетесь от подписки, вы получите тот результат, который ищете.

Пример длительной отписки от подписки здесь.