@Override
    public void run() {
        logger.info(this.getServiceName() + " service started");

        while (!this.isStopped()) {
            try {
                MessageRequest messageRequest = this.messageRequestQueue.take();
                if (messageRequest.getMessageRequestMode() == MessageRequestMode.POP) {
                    this.popMessage((PopRequest) messageRequest);
                } else {
                    this.pullMessage((PullRequest) messageRequest);
                }
            } catch (InterruptedException ignored) {
            } catch (Exception e) {
                logger.error("Pull Message Service Run Method exception", e);
            }
        }

        logger.info(this.getServiceName() + " service end");
    }
@Test
    public void testRun() throws InterruptedException, IllegalAccessException {
        LinkedBlockingQueue<MessageRequest> messageRequestQueue = new LinkedBlockingQueue<>();
        PopRequest popRequest = mock(PopRequest.class);
        when(popRequest.getMessageRequestMode()).thenReturn(MessageRequestMode.POP);
        when(popRequest.getConsumerGroup()).thenReturn(defaultGroup);
        messageRequestQueue.put(popRequest);
        DefaultMQPushConsumerImpl defaultMQPushConsumerImpl = mock(DefaultMQPushConsumerImpl.class);
        when(mQClientFactory.selectConsumer(any())).thenReturn(defaultMQPushConsumerImpl);
        FieldUtils.writeDeclaredField(pullMessageService, "messageRequestQueue", messageRequestQueue, true);
        new Thread(() -> pullMessageService.run()).start();
        TimeUnit.SECONDS.sleep(1);
        pullMessageService.makeStop();
        verify(mQClientFactory, times(1)).selectConsumer(eq(defaultGroup));
        verify(defaultMQPushConsumerImpl).popMessage(any(PopRequest.class));
    }