Java gRPC双向流服务对接ActiveMQ时JmsTemplate为空问题咨询

• Spring托管Bean手动实例化导致依赖注入失效：你在TransferHelloRequestStreaming的onNext方法中直接通过new HelloProducer()创建实例，@Service、@Autowired注解只有在对象由Spring容器管理时才会生效，手动创建的对象不会被Spring注入依赖，因此JmsTemplate始终为null。
• TransferHelloRequestStreaming实例化方式错误：你在HelloService的transfer方法中手动new了TransferHelloRequestStreaming实例，哪怕该类加了@Service注解，手动创建的实例也不受Spring容器管理，后续注入其他Bean同样会失效。
• 配置笔误：application.yml中spring.application-name配置项格式错误，正确格式为spring.application.name，该问题不影响ActiveMQ功能，仅影响服务名识别。
• 方法入参类型不匹配：HelloProducer的sendTo方法第二个参数类型为MessageEnvelope，但调用时传入的是HelloEnvelope，类型不匹配会触发运行时异常。

1. 修正TransferHelloRequestStreaming实例化逻辑 #

将TransferHelloRequestStreaming设置为原型作用域，每次gRPC流请求从Spring容器获取实例，同时注入HelloProducer依赖：

@Slf4j
@Service
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE) // 每次调用生成新实例，适配gRPC流的多请求场景
public class TransferHelloRequestStreaming implements StreamObserver<HelloEnvelope> {

    private StreamObserver<HelloEnvelope> transferHelloResponseStreamObserver;
    // 注入Spring管理的HelloProducer
    @Autowired
    private HelloProducer helloProducer;

    public TransferHelloRequestStreaming(StreamObserver<HelloEnvelope> transferHelloResponseStreamObserver) {
        this.transferHelloResponseStreamObserver = transferHelloResponseStreamObserver;
    }

    @Override
    public void onNext(HelloEnvelope transferHelloRequest) {
        // 使用注入的实例，不要手动new
        helloProducer.sendTo("input", transferHelloRequest);
    }

    @Override
    public void onError(Throwable throwable) {
        log.error("gRPC流处理异常", throwable);
    }

    @Override
    public void onCompleted() {
        this.transferHelloResponseStreamObserver.onCompleted();
    }
}

修改HelloService，从Spring上下文获取TransferHelloRequestStreaming实例：

@GrpcService
public class HelloService extends HelloServiceGrpc.HelloServiceImplBase {
    // 注入Spring上下文
    @Autowired
    private ApplicationContext applicationContext;

    @Override
    public StreamObserver<HelloEnvelope> transfer(StreamObserver<HelloEnvelope> responseObserver) {
        // 从Spring容器获取实例，传入responseObserver构造参数
        return applicationContext.getBean(TransferHelloRequestStreaming.class, responseObserver);
    }
}

2. 显式配置JmsTemplate（可选，避免自动配置失效） #

Spring Boot ActiveMQ启动器默认会自动装配JmsTemplate，如果自动装配失效可以在JmsConfig中手动配置：

@Configuration
@EnableJms
public class JmsConfig {

    @Bean
    public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(
            ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setConcurrency("5-10");
        return factory;
    }

    @Bean
    public JmsTemplate jmsTemplate(ConnectionFactory connectionFactory) {
        JmsTemplate jmsTemplate = new JmsTemplate(connectionFactory);
        // 队列模式设为false，主题模式设为true，根据业务需求调整
        jmsTemplate.setPubSubDomain(false);
        return jmsTemplate;
    }
}

3. 修正入参类型不匹配问题 #

修改HelloProducer的sendTo方法参数类型，和传入的HelloEnvelope保持一致：

@Service
@Slf4j
public class HelloProducer {

  @Autowired
  JmsTemplate jmsTemplate;

  // 修正参数类型为HelloEnvelope
  public void sendTo(String destination, HelloEnvelope transferHelloRequest) {
    jmsTemplate.convertAndSend(destination, transferHelloRequest);
  }
}

4. 修正application.yml配置笔误 #

spring:
  application:
    name: bidirectional-server
  activemq:
    broker-url: tcp://localhost:61616
    user: admin
    password: admin
    packages.trust-all: true

1. 确认ActiveMQ服务正常启动，61616端口可正常访问
2. 启动项目后在HelloProducer的sendTo方法打断点，验证jmsTemplate已被正常注入不为null
3. 调用gRPC接口测试消息是否可以正常推送到ActiveMQ对应队列

内容的提问来源于stack exchange，提问作者Akhilesh Pandey