使用Spring Data R2DBC和Reactive MySQL Driver映射到JSON字段

Question

我正在尝试使用JSON数据类型在MySQL数据库的特定列中存储JSON数据。

我遵循这个指南，并得到存储和读取JSON工作良好，但现在我得到这个错误时，试图保存另一个实体有一个boolean列/字段:

Caused by: org.springframework.core.convert.ConverterNotFoundException: No converter found capable of converting from type [java.lang.Byte] to type [boolean]
at org.springframework.core.convert.support.GenericConversionService.handleConverterNotFound(GenericConversionService.java:322)
at org.springframework.core.convert.support.GenericConversionService.convert(GenericConversionService.java:195)
at org.springframework.core.convert.support.GenericConversionService.convert(GenericConversionService.java:175)
at org.springframework.data.r2dbc.convert.MappingR2dbcConverter.getPotentiallyConvertedSimpleRead(MappingR2dbcConverter.java:277)
at org.springframework.data.r2dbc.convert.MappingR2dbcConverter.readValue(MappingR2dbcConverter.java:201)
at org.springframework.data.r2dbc.convert.MappingR2dbcConverter.readFrom(MappingR2dbcConverter.java:182)
at org.springframework.data.r2dbc.convert.MappingR2dbcConverter.read(MappingR2dbcConverter.java:136)
at org.springframework.data.r2dbc.convert.MappingR2dbcConverter.read(MappingR2dbcConverter.java:119)
at org.springframework.data.r2dbc.convert.EntityRowMapper.apply(EntityRowMapper.java:46)
at org.springframework.data.r2dbc.convert.EntityRowMapper.apply(EntityRowMapper.java:29)
at dev.miku.r2dbc.mysql.MySqlResult.processRow(MySqlResult.java:176)
at dev.miku.r2dbc.mysql.MySqlResult.handleResult(MySqlResult.java:149)
at dev.miku.r2dbc.mysql.MySqlResult.lambda$map$1(MySqlResult.java:93)
at reactor.core.publisher.FluxHandle$HandleSubscriber.onNext(FluxHandle.java:102)
at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.onNext(FluxContextWrite.java:107)
at dev.miku.r2dbc.mysql.util.DiscardOnCancelSubscriber.onNext(DiscardOnCancelSubscriber.java:70)
at reactor.core.publisher.FluxWindowPredicate$WindowFlux.drainRegular(FluxWindowPredicate.java:667)
at reactor.core.publisher.FluxWindowPredicate$WindowFlux.drain(FluxWindowPredicate.java:745)
at reactor.core.publisher.FluxWindowPredicate$WindowFlux.onNext(FluxWindowPredicate.java:787)
at reactor.core.publisher.FluxWindowPredicate$WindowPredicateMain.onNext(FluxWindowPredicate.java:265)
at reactor.core.publisher.FluxHandleFuseable$HandleFuseableSubscriber.onNext(FluxHandleFuseable.java:184)
at reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber.onNext(FluxContextWrite.java:107)
at dev.miku.r2dbc.mysql.util.DiscardOnCancelSubscriber.onNext(DiscardOnCancelSubscriber.java:70)
at reactor.core.publisher.FluxPeekFuseable$PeekConditionalSubscriber.onNext(FluxPeekFuseable.java:854)
at reactor.core.publisher.MonoFlatMapMany$FlatMapManyInner.onNext(MonoFlatMapMany.java:250)
at reactor.core.publisher.FluxPeek$PeekSubscriber.onNext(FluxPeek.java:199)
at reactor.core.publisher.FluxHandle$HandleSubscriber.onNext(FluxHandle.java:118)
at reactor.core.publisher.FluxPeekFuseable$PeekConditionalSubscriber.onNext(FluxPeekFuseable.java:854)
at reactor.core.publisher.EmitterProcessor.drain(EmitterProcessor.java:491)
at reactor.core.publisher.EmitterProcessor.tryEmitNext(EmitterProcessor.java:299)
at reactor.core.publisher.InternalManySink.emitNext(InternalManySink.java:27)
at reactor.core.publisher.EmitterProcessor.onNext(EmitterProcessor.java:265)
at dev.miku.r2dbc.mysql.client.ReactorNettyClient$ResponseSink.next(ReactorNettyClient.java:340)
at dev.miku.r2dbc.mysql.client.ReactorNettyClient.lambda$new$0(ReactorNettyClient.java:103)
at reactor.core.publisher.FluxPeek$PeekSubscriber.onNext(FluxPeek.java:184)
at reactor.netty.channel.FluxReceive.drainReceiver(FluxReceive.java:280)
at reactor.netty.channel.FluxReceive.onInboundNext(FluxReceive.java:389)
at reactor.netty.channel.ChannelOperations.onInboundNext(ChannelOperations.java:405)
at reactor.netty.channel.ChannelOperationsHandler.channelRead(ChannelOperationsHandler.java:94)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at dev.miku.r2dbc.mysql.client.MessageDuplexCodec.handleDecoded(MessageDuplexCodec.java:187)
at dev.miku.r2dbc.mysql.client.MessageDuplexCodec.channelRead(MessageDuplexCodec.java:95)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:324)
at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:311)
at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:432)
at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1368)
at io.netty.handler.ssl.SslHandler.decodeJdkCompatible(SslHandler.java:1234)
at io.netty.handler.ssl.SslHandler.decode(SslHandler.java:1280)
at io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:507)
at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:446)
at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:357)
at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:365)
at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:719)
at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:655)
at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:581)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)
at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.base/java.lang.Thread.run(Thread.java:834)

如果我不创建R2dbcCustomConversionsbean(如指南中所述)，那么我不会得到这个错误，但JSON不工作…

是否有另一种方法在我的场景中使用JSON ?

Answer 1

回答我自己的问题-看起来像是spring-data-r2dbc中的某种错误:

在R2dbcDataAutoConfiguration中有这个bean定义:

@Bean
@ConditionalOnMissingBean
public R2dbcCustomConversions r2dbcCustomConversions() {
List<Object> converters = new ArrayList<>(this.dialect.getConverters());
converters.addAll(R2dbcCustomConversions.STORE_CONVERTERS);
return new R2dbcCustomConversions(
CustomConversions.StoreConversions.of(this.dialect.getSimpleTypeHolder(), converters),
Collections.emptyList());
}

这是我们在注册转换器时重写的bean(这是我的代码):

@Bean
public R2dbcCustomConversions r2dbcCustomConversions(JsonEventsReadingConverter readingConverter,
JsonEventsWritingConverter writingConverter) {
val converters = new ArrayList<>();
converters.add(readingConverter);
converters.add(writingConverter);
return new R2dbcCustomConversions(converters);
}

CTORnew R2dbcCustomConversions(converters)负责添加一些基本转换器(STORE_CONVERSIONS&R2dbcConverters)，但没有添加MySqlDialect.BooleanToByteConverter&MySqlDialect.ByteToBooleanConverter负责MySQL值到Java值的转换。

我找到了一个棘手的解决方案(手动在我的bean中添加那些转换器)，但我没有看到其他方法(开放建议):

@Configuration
@RequiredArgsConstructor
public class R2dbcConvertorsConfig {
private final DatabaseClient databaseClient;
@Bean
public R2dbcCustomConversions r2dbcCustomConversions(JsonEventsReadingConverter readingConverter,
JsonEventsWritingConverter writingConverter) {
val dialect = DialectResolver.getDialect(this.databaseClient.getConnectionFactory());
val converters = new ArrayList<>();
converters.add(readingConverter);
converters.add(writingConverter);
converters.addAll(dialect.getConverters());
return new R2dbcCustomConversions(converters);
}
}

要求:

package com.mycompany.platform.webhookdelivery.core.dal.converters;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.mycompany.platform.webhookdelivery.core.dal.entities.webhookfilters.JsonEvents;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.convert.converter.Converter;
import org.springframework.core.serializer.support.SerializationFailedException;
import org.springframework.data.convert.ReadingConverter;
import org.springframework.lang.NonNull;
import org.springframework.stereotype.Component;
import java.io.IOException;
@Slf4j
@ReadingConverter
@AllArgsConstructor
@Component
public class JsonEventsReadingConverter implements Converter<String, JsonEvents> {
private final ObjectMapper objectMapper;
@Override
public JsonEvents convert(@NonNull String target) {
try {
String json = " {"events":" + target + "}";
return objectMapper.readValue(json, JsonEvents.class);
} catch (IOException e) {
log.debug("Failed to deserialise  JSON: " + target, e);
throw new SerializationFailedException("Failed to deserialize  JSON: " + target, e);
}
}
}

package com.mycompany.platform.webhookdelivery.core.dal.converters;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.mycompany.platform.webhookdelivery.core.dal.entities.webhookfilters.JsonEvents;
import lombok.AllArgsConstructor;
import lombok.extern.log4j.Log4j2;
import org.springframework.core.convert.converter.Converter;
import org.springframework.core.serializer.support.SerializationFailedException;
import org.springframework.data.convert.WritingConverter;
import org.springframework.stereotype.Component;
@Log4j2
@WritingConverter
@AllArgsConstructor
@Component
public class JsonEventsWritingConverter implements Converter<JsonEvents, String> {
private final ObjectMapper objectMapper;
@Override
public String convert(JsonEvents source) {
try {
return objectMapper.writeValueAsString(source.getEvents());
} catch (JsonProcessingException e) {
log.debug("Failed to serialise JSON: " + source, e);
throw new SerializationFailedException("Failed to serialize JSON: " + source, e);
}
}
}