小编Yas*_*lan的帖子

我正在尝试使用 R2DBC 编写一些反应式后端，但遇到了我不太理解的事情。我不知道为什么，但 Spring 正在尝试更新条目而不是将其保存到数据库中。

我的用户模型：

@Table("user")
@NoArgsConstructor
@Data
public class User {

    @Id
    private String id;

    private String username;
    private String password;

    public User(String id,String username,String password){
        this.id = id;
        this.username = username;
        this.password=password;
    }

我的控制器：

@RestController
    public class UserController {

`   private final UserRepository userRepository`;

    @Autowired
    public UserController(UserRepository userRepository){

        this.userRepository = userRepository;
    }

    @PostMapping("/create")
        public Mono<User> createUser(@RequestBody User user){
            return userRepository.save(user);
        }   

当我使用如下输入向此端点发送 POST 请求时：

{“id”：“re”，“用户名”：“ehmmidk”，“密码”：“dsadsadsadsa”}

即使我的数据库中没有条目，它也会进行更新而不是保存。

我的架构如下所示：

CREATE TABLE user (
   id VARCHAR(255) NOT NULL ,
   username …

现在，我在我的 SpringBoot 应用程序中实现了一个简单的 Kafka Consumer 和 Producer，它工作得很好，接下来我想做的是，我的消费者获取消费的消息并将其直接广播给所有订阅的客户端。我发现我不能将 STOMP 消息传递与 WebFlux 一起使用，那么我该如何完成这个任务，我看到了反应式 WebSocket 实现，但我不知道如何将我使用的数据发送到我的 websocket。

这是我的简单 KafkaProducer：

fun addMessage(message: Message){
        val headers : MutableMap<String, Any> = HashMap()
        headers[KafkaHeaders.TOPIC] = topicName
        kafkaTemplate.send(GenericMessage<Message>(message, headers))
    }

我的简单消费者看起来像这样：

@KafkaListener(topics = ["mytopic"], groupId = "test-consumer-group")
    fun receiveData(message:Message) :Message{
        //Take consumed data and send to websocket
    }