@Override
        void run() {
            try {
                Runtime.getRuntime().addShutdownHook(new Thread(() -> {
                    if (msgTotal == -1)
                        stopProducer(msgCount.get());
                }));

                final TubeClientConfig clientConfig = new TubeClientConfig(masterServers);
                final TubeSingleSessionFactory messageSessionFactory = new TubeSingleSessionFactory(clientConfig);
                messageProducer = messageSessionFactory.createProducer();
                messageProducer.publish(topicName);
                byte[] bodyData;
                final Message message = new Message(topicName, null);

                BufferedReader input = new BufferedReader(new InputStreamReader(System.in));
                int c = 0;
                while (msgTotal == -1 || c < msgTotal) {
                    System.out.print(">");
                    body = input.readLine();
                    if (body == null || "".equals(body) || "".equals(body.trim()))
                        continue;
                    bodyData = StringUtils.getBytesUtf8(body);
                    message.setData(bodyData);

                    switch (mode) {
                        case "sync":
                            syncProduce(message);
                            break;
                        case "async":
                            asyncProduce(message);
                            break;
                        default:
                            throw new ParameterException("Produce mode, must in { sync | async }");
                    }
                    c++;
                }
                stopProducer(msgTotal);
            } catch (Exception e) {
                System.out.println(e.getMessage());
            } catch (Throwable e) {
                e.printStackTrace();
            }
        }
Stop a producer and print the total number of messages produced

@param v  total number of messages
@throws TubeClientException
@throws InterruptedException
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/tools/cli/MessageCommand.java