RabbitMQ Spring Boot 会话错误

Question

我在尝试将 RabbitMQ 队列消息传递给我的 Spring Boot 应用程序时遇到了问题。我正在运行 spring data 4 并使用 Neo4j 作为我的数据库。我能够从 RabbitMQ 队列接收 JSON 字符串并将其解析到 Spring Boot 应用程序。这是接收进程的线程更改：

14:19:17.811 [main] INFO o.s.b.c.e.t.TomcatEmbeddedServletContainer - Tomcat started on port(s): 8080 (http) 14:19:17.813 [main] INFO org.directoryx.Application - Started Application in 4.989 seconds (JVM running for 5.47) 14:19:26.708 [SimpleAsyncTaskExecutor-1] INFO org.directoryx.sync.Receiver - Message Recv'd! {"lastName":"Okuneva", "country":"United States", "city":"New Stantonfort","org":"IT","photo":"profile9.jpg", "active":"true", "managerId":"1502", "title":"Global Integration Orchestrator", "userId":"frokune", "firstName":"Frank", "phone":"(485)544-1782 x913", "id":"6791", "email":"frokune@peoplemaker.com","status":"Employee"} 14:19:26.718 [SimpleAsyncTaskExecutor-1] INFO org.directoryx.sync.Receiver - Update Person: Frank Okuneva

但是当我尝试将传入的 JSON 保存到 Neo4j 节点时，应用程序崩溃并出现以下错误：

Exception in thread "main" org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'receiver': Injection of autowired dependencies failed; nested exception is org.springframework.beans.factory.BeanCreationException: Could not autowire field: private org.springframework.data.neo4j.template.Neo4jTemplate

在实现 neo4jTemplate 之前，当我尝试通过 personService->personServiceImpl->personRepository 模式保存人员时遇到此错误。

Caused by: org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'scopedTarget.getSession': Scope 'session' is not active for the current thread; consider defining a scoped proxy for this bean if you intend to refer to it from a singleton; nested exception is java.lang.IllegalStateException: No thread-bound request found: Are you referring to request attributes outside of an actual web request, or processing a request outside of the originally receiving thread? If you are actually operating within a web request and still receive this message, your code is probably running outside of DispatcherServlet/DispatcherPortlet: In this case, use RequestContextListener or RequestContextFilter to expose the current request.

以下是涉及的三个类：

Receiver.java

@Service
public class Receiver implements MessageListener {

static Logger log = LoggerFactory.getLogger(Receiver.class);

// Fails to be injected...
// Injection of autowired dependencies failed; nested exception is
// org.springframework.beans.factory.BeanCreationException: Could not autowire field:
@Autowired
private Neo4jTemplate neo4jTemplate;

public void onMessage(Message message) {
    Gson gson  = new GsonBuilder().create();
    String msg = new String(message.getBody());
    log.info("Message Recv'd! {}",msg);

    Person person = gson.fromJson(msg, Person.class);
    log.info("Update Person: {} {}", person.getFirstName(), person.getLastName());

    try {
        neo4jTemplate.save(person);
    } catch (Exception e) {
        log.error("Receiver failed to save Person: {}", e);
    }
}

DataSyncConfig.java

@Configuration
public class DataSyncConfig {


@Autowired
RabbitTemplate rabbitTemplate;
private final String QUEUE_NAME = "sync2";
private final String ROUTING_KEY = "routeme";
private final boolean DURABLE = true;
private final boolean EXCLUSIVE = true;
private final boolean AUTO_DELETE = true;

@Bean
Queue queue() {
    return new Queue(QUEUE_NAME, DURABLE, !EXCLUSIVE, !AUTO_DELETE);
}

@Bean
TopicExchange exchange() {
    return new TopicExchange("etl-sync-tool");
}

@Bean
Binding binding(Queue queue, TopicExchange exchange) {
    return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
}

@Bean
public MessageConverter jsonMessageConverter() {
    JsonMessageConverter jsonMessageConverter = new JsonMessageConverter();
    return jsonMessageConverter;
}

@Bean
SimpleMessageListenerContainer container(ConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {
    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
    container.setConnectionFactory(connectionFactory);
    container.setQueueNames(QUEUE_NAME);
    container.setMessageListener(listenerAdapter);
    return container;
}

@Bean
MessageListenerAdapter listenerAdapter(Receiver receiver) {
    return new MessageListenerAdapter(receiver);
}
}

Neo4jConfig.java

@Configuration
@EnableNeo4jRepositories(basePackages = "org.directoryx.repository", queryLookupStrategy = QueryLookupStrategy.Key.CREATE_IF_NOT_FOUND)
@EnableTransactionManagement
public class Neo4jConfig extends Neo4jConfiguration {

private final Logger log = LoggerFactory.getLogger(Neo4jConfig.class);

@Resource
public Environment env;

@Override
@Bean
public Neo4jServer neo4jServer() {
    log.info("Initialising Neo4j server connection...");

    // username and password need to be available as System properties for Neo4j authentication.
    String user = env.getProperty("neo4j.user");
    System.setProperty("username", user);

    String pass = env.getProperty("neo4j.pass");
    System.setProperty("password", pass);

    log.info("connecting as neo4j.user=" + user);
    return new RemoteServer("http://localhost:7474");
}

@Override
@Bean
public SessionFactory getSessionFactory() {
    log.info("Initialising Session Factory");
    return new SessionFactory("org.directoryx.domain");
}

@Override
@Bean
// Unsure of scope value, tried both "session" and "global"
@Scope(value = "session", proxyMode = ScopedProxyMode.TARGET_CLASS)
public Session getSession() throws Exception {
    log.info("Initialising session-scoped Session Bean");
    return super.getSession();
}

@Bean
public Neo4jTemplate neo4jTemplate (Session session ) {
    return new Neo4jTemplate(session);
}
}

我似乎不知道如何通过公开和使用当前线程或创建新线程来将数据保存到数据库中。我觉得这是一个与 Spring Boot 和 Spring Data 的一些核心功能相距不远的问题，但即使经过大量研究，我似乎也无法确定要使用什么语法。

Answer 1

错误消息表明使用会话范围运行不适合您的使用上下文，可能是因为您处于 RabbitMQ 上下文而不是 Spring Web/MVC 上下文中。

我建议从Neo4jConfig 中的getSession 方法中完全删除@Scope 注释，看看这是否会有所改善。如果您认为有必要，您可以随时添加用户管理的范围。

此外，您可能需要考虑针对Neo4jOperations 而不是Neo4jTemplate 进行编码。一些用户报告了注入类而不是接口的问题。

Answer 2

我删除了@Scope 注释块并保留了覆盖的 SessionFactory。

Neoj4Config.java

@Configuration
@EnableNeo4jRepositories(basePackages = "org.directoryx.repository",   queryLookupStrategy = QueryLookupStrategy.Key.CREATE_IF_NOT_FOUND)
@EnableTransactionManagement
public class Neo4jConfig extends Neo4jConfiguration {

private final Logger log = LoggerFactory.getLogger(Neo4jConfig.class);

@Resource
public Environment env;

@Override
@Bean
public Neo4jServer neo4jServer() {
    log.info("Initialising Neo4j server connection...");

    // username and password need to be available as System properties for Neo4j authentication.
    String user = env.getProperty("neo4j.user");
    System.setProperty("username", user);

    String pass = env.getProperty("neo4j.pass");
    System.setProperty("password", pass);

    log.info("connecting as neo4j.user=" + user);
    return new RemoteServer("http://localhost:7474");
}

@Override
@Bean
public SessionFactory getSessionFactory() {
    log.info("Initialising Session Factory");
    return new SessionFactory("org.directoryx.domain");
}
}