Очередь сообщений

<mq name="mymq" type="kafka">
    <servers value="127.0.0.1:9092"/>
    <consumer autoload="true"/>
</mq>

@ResourceConsumer(mq = "mymq", topics = "test_bean_topic")
public class TestMessageConsumer implements MessageConsumer<TestBean> {

    @Override
    public void init(AnyValue config) {
        System.out.println("Выполнение TestMessageConsumer.init");
    }

    @Override
    public void onMessage(MessageEvent<TestBean>[] events) {
        for (MessageEvent<TestBean> event : events) {
            System.out.println("Потребление сообщения, сообщение: " + event.getMessage());
        }
    }

    @Override
    public void destroy(AnyValue config) {
        System.out.println("Выполнение TestMessageConsumer.destroy");
    }
}

public class TestMessageService extends AbstractService {

    @Messaged(mq = "mymq", topics = "test_bean_topic")
    protected void runMessage(MessageEvent<TestBean>[] events) {
        for (MessageEvent<TestBean> event : events) {
            System.out.println("Потребление сообщения, сообщение: " + event.getMessage());
        }
    }
}

@Component
public final class TestMessageService extends AbstractService {

    @Messaged(mq = "mymq", topics = "test_bean_topic")
    public int runMessage(MessageEvent<TestBean>[] events) {
        for (MessageEvent<TestBean> event : events) {
            System.out.println("Потребление сообщения, сообщение: " + event.getMessage());
        }
        return 0;
    }
}

@Component
public class TestMessageService extends AbstractService {

    @ResourceProducer(mq = "mymq")
    private MessageProducer producer;

    public void sendMessage() {
        TestBean bean = new TestBean(12345, "this is a message");
        System.out.println("Создание сообщения: " + bean);
        producer.sendMessage("test_bean_topic", bean);
    }
}

public class TestMessageManager extends AbstractService {

    @Resource(name = "mymq")
    private MessageManager manager;

    // Создание темы
    public void initTopic() {
        manager.createTopic("topic_1", "topic_2").join();
    }

    // Удаление темы
    public void deleteTopic() {
        manager.deleteTopic("topic_1", "topic_2").join();
    }

    // Запрос темы
    public void printTopic() {
        List<String> topics = manager.queryTopic().join();
    }
}