java - 如何使用 Pulsar 作为 Log4j2 的 Appender？

Question

我正在尝试将 Apache Pulsar 设置为 log4j2 的附加程序。它的文档不多，但我设法在这里找到了一些示例。

我设置了一个玩具示例，其中生产者记录一条消息，消费者订阅 Pulsar 服务器上的同一主题并监听消息。消费者收到了一些东西，但不是我期待的消息:

Message sent: Test message
Message received: 18:05:27.510 [pulsar-client-io-5-1] INFO  org.apache.pulsar.client.impl.ConsumerImpl
Message received: 18:05:27.510 [main] INFO  com.pulsar_logging.Producer

Producer.java:

package com.pulsar_logging;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.pulsar.client.api.PulsarClientException;

public class Producer {

    private static final Logger logger = LogManager.getLogger(Producer.class);

    public static void main(String[] args) {
        Consumer consumer = null;
        try {
            consumer = new Consumer("my-topic", "my-subscription");
        } catch (PulsarClientException e) {
            e.printStackTrace();
        }

        String msg = "Test message";
        System.out.printf("Message sent: %s\n", msg);
        logger.info(msg.getBytes());

        try {
            consumer.listen();
        } catch (PulsarClientException e) {
            e.printStackTrace();
        }
    }
}

消费者.java:

package com.pulsar_logging;

import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;

public class Consumer {

    org.apache.pulsar.client.api.Consumer<byte[]> consumer;
    PulsarClient client;
    private final String SERVICE_URL = "pulsar://localhost:6650";

    public Consumer(String topic, String subName) throws PulsarClientException {
        client = PulsarClient.builder()
                .serviceUrl(SERVICE_URL)
                .build();
        consumer = client.newConsumer()
                .topic(topic)
                .subscriptionName(subName)
                .subscribe();
    }

    public void listen() throws PulsarClientException {
        while (true) {
            Message msg = consumer.receive();

            try {
                System.out.printf("Message received: %s\n", new String(msg.getData()));
                consumer.acknowledge(msg);
            } catch (Exception e) {
                System.out.println("MESSAGE FAILED");
                consumer.negativeAcknowledge(msg);
            }
        }

    }
}

log4j2.xml:

<Configuration status="WARN">
    <Properties>
        <Property name="DefaultPattern" value="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36}"/>
    </Properties>
    <Appenders>
        <Pulsar name="Pulsar_Appender" serviceUrl="pulsar://localhost:6650" topic="my-topic" avoidRecursive="false">
            <PatternLayout pattern="${DefaultPattern}"/>
        </Pulsar>
    </Appenders>
    <Loggers>
        <Logger name="org.apache.pulsar" level="INFO"/>
        <Root level="debug">
            <AppenderRef ref="Pulsar_Appender"/>
        </Root>
    </Loggers>
</Configuration>

Answer 1

您从未说过您的期望，但它似乎完全按照您的要求去做。

您已经声明了两个记录器 - 信息级别的 org.apache.pulsar 和调试级别的根记录器。两者都将使用 Pulsar appender，其布局记录小时(包括毫秒)、线程名称、日志记录级别和记录器名称。您没有指定应该包含该消息，所以它没有。

我在发送消息时看到一个来自 Pulsar 的日志，另一个来自你的测试应用程序，它记录了消息。