之前在网上找了个例子：https://www.freesion.com/article/53297012/ 写的挺好。
github已经有开源地址RocketMQ-Flink。（读取可以根据github和这个例子写）
上面那个例子写了String的反序列化类SimpleStringDeserializationSchema。没有写对应的序列化类。
1、今天写项目就恰好用到了。记录下：
import org.apache.rocketmq.flink.common.serialization.KeyValueSerializationSchema;
import java.nio.charset.StandardCharsets;
public class SimpleStringSerializationSchema implements KeyValueSerializationSchema<String> {
private static final long serialVersionUID = 609645086382422113L;
@Override
public byte[] serializeKey(String key) {
return null;
}
@Override
public byte[] serializeValue(String value) {
return value != null ? value.getBytes(StandardCharsets.UTF_8) : null;
}
}
注意：serialVersionUID 这个是用idea生成的。在自己包下重新生成下，以免有问题。（方法：https://blog.csdn.net/ShotMoon/article/details/80496407）
2、写入MQ:
Properties props1 = new Properties();
props1.setProperty(RocketMQConfig.NAME_SERVER_ADDR, Config.getMQServer());
props1.setProperty(RocketMQConfig.PRODUCER_GROUP,Config.getProduceGroup());
DefaultTopicSelector selector = new DefaultTopicSelector(Config.getTopic(),Config.getTagName());
KeyValueSerializationSchema schema1 = new SimpleStringSerializationSchema();
result.addSink(new RocketMQSink(schema1, selector, props1));
具体参数按照自己的写就行。
完美。
