1.1DataSink数据输出
经过一系列Transformation转换操作后，最后一定要调用Sink操作，才会形成一个完整的DataFlow拓扑。只有调用了Sink操作，才会产生最终的计算结果，这些数据可以写入到的文件、输出到指定的网络端口、消息中间件、外部的文件系统或者是打印到控制台。
1.1.1print 打印
打印是最简单的一个Sink，通常是用来做实验和测试时使用。如果想让一个DataStream输出打印的结果，直接可以在该DataStream调用print方法。另外，该方法还有一个重载的方法，可以传入一个字符，指定一个Sink的标识名称，如果有多个打印的Sink，用来区分到底是哪一个Sink的输出。
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.util.Collector;
public class PrintSinkDemo {
public static void main(String[] args) throws Exception {
//local模式默认的并行度是当前机器的逻辑核的数量
Configuration configuration = new Configuration();
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
int parallelism0 = env.getParallelism();
System.out.println("执行环境默认的并行度：" + parallelism0);
DataStreamSource<String> lines = env.socketTextStream("cs-28-86", 8888);
//获取DataStream的并行度
int parallelism = lines.getParallelism();
System.out.println("SocketSource的并行度：" + parallelism);
lines.print();
//lines.addSink(new MyPrintSink()).name("my-print-sink");
env.execute();
}
public static class MyPrintSink extends RichSinkFunction<String> {
private int indexOfThisSubtask;
@Override
public void open(Configuration parameters) throws Exception {
RuntimeContext runtimeContext = getRuntimeContext();
indexOfThisSubtask = runtimeContext.getIndexOfThisSubtask();
}
@Override
public void invoke(String value, Context context) throws Exception {
System.out.println(indexOfThisSubtask + 1 + "> " + value);
}
}
}
下面的结果是WordCount例子中调用print Sink输出在控制台的结果，细心的读者会发现，在输出的单词和次数之前，有一个数字前缀，我这里是1~4，这个数字是该Sink所在subtask的Index + 1。有的读者运行的结果数字前缀是1~8，该数字前缀其实是与任务的并行度相关的，由于该任务是以local模式运行，默认的并行度是所在机器可用的逻辑核数即线程数，我的电脑是2核4线程的，所以subtask的Index范围是0~3，将Index + 1，显示的数字前缀就是1~4了。这里在来仔细的观察一下运行的结果发现：相同的单词输出结果的数字前缀一定相同，即经过keyBy之后，相同的单词会被shuffle到同一个subtask中，并且在同一个subtask的同一个组内进行聚合。一个subtask中是可能有零到多个组的，如果是有多个组，每一个组是相互独立的，累加的结果不会相互干扰。
1.1.2writerAsText 以文本格式输出
该方法是将数据以文本格式实时的写入到指定的目录中，本质上使用的是TextOutputFormat格式写入的。每输出一个元素，在该内容后面同时追加一个换行符，最终以字符的形式写入到文件中，目录中的文件名称是该Sink所在subtask的Index + 1。该方法还有一个重载的方法，可以额外指定一个枚举类型的参数writeMode，默认是WriteMode.NO_OVERWRITE，如果指定相同输出目录下有相同的名称文件存在，就会出现异常。如果是WriteMode.OVERWRITE，会将以前的文件覆盖。
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
public class WriteSinkDemo {
public static void main(String[] args) throws Exception {
//local模式默认的并行度是当前机器的逻辑核的数量
Configuration configuration = new Configuration();
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
int parallelism0 = env.getParallelism();
System.out.println("执行环境默认的并行度：" + parallelism0);
DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);
//获取DataStream的并行度
int parallelism = lines.getParallelism();
System.out.println("SocketSource的并行度：" + parallelism);
lines.writeAsText("file:///Users/xing/Desktop/out");
env.execute();
}
public static class MyPrintSink extends RichSinkFunction<String> {
private int indexOfThisSubtask;
@Override
public void open(Configuration parameters) throws Exception {
RuntimeContext runtimeContext = getRuntimeContext();
indexOfThisSubtask = runtimeContext.getIndexOfThisSubtask();
}
@Override
public void invoke(String value, Context context) throws Exception {
System.out.println(indexOfThisSubtask + 1 + "> " + value);
}
}
}
1.1.3writeAsCsv 以csv格式输出
该方法是将数据以csv格式写入到指定的目录中，本质上使用的是CsvOutputFormat格式写入的。每输出一个元素，在该内容后面同时追加一个换行符，最终以csv的形式(类似Excel的格式，字段和字段之间用逗号分隔)写入到文件中，目录中的文件名称是该Sink所在subtask的Index + 1。需要说明的是，该Sink并不是将数据实时的写入到文件中，而是有一个BufferedOutputStream，默认缓存的大小为4096个字节，只有达到这个大小，才会flush到磁盘。另外程序在正常退出，调用Sink的close方法也会flush到磁盘。
DataStream<Tuple2<String, Integer>> result = wordAndOne.keyBy(0).sum(1);1.1.4writeUsingOutputFormat以指定的格式输出
result.writeAsCsv(path);
该方法是将数据已指定的格式写入到指定目录中，该方法要传入一个OutputFormat接口的实现类，该接口有很多已经实现好了的实现类，并且可以根据需求自己实现，所以该方法更加灵活。writeAsText和writeAsCsv方法底层都是调用了writeUsingOutputFormat方法。
DataStream<Tuple2<String, Integer>> result = wordAndOne.keyBy(0).sum(1);1.1.5writeToSocket输出到网络端口
result.writeUsingOutputFormat(new TextOutputFormat<>(new Path(path));
该方法是将数据输出到指定的Socket网络地址端口。该方法需要传入三个参数：第一个为ip地址或主机名，第二个为端口号，第三个为数据输出的序列化格式SerializationSchema。输出之前，指定的网络端口服务必须已经启动。
DataStreamSource<String> lines = env.socketTextStream(“localhost”, 8888);1.1.6RedisSink
lines.writeToSocket(“localhost”, 9999, new SimpleStringSchema());
该方法是将数据输出到Redis数据库中，Redis是一个基于内存、性能极高的NoSQL数据库，数据还可以持久化到磁盘，读写速度快，适合存储key-value类型的数据。Redis不仅仅支持简单的key-value类型的数据，同时还提供list，set，zset，hash等数据结构的存储。Flink实时计算出的结果，需要快速的输出存储起来，要求写入的存储系统的速度要快，这个才不会造成数据积压。Redis就是一个非常不错的选择。
首先在maven项目中的pom.xml中添加Redis Sink的依赖。
<!-- redis 依赖-->
<dependency>
<groupId>org.apache.bahir</groupId>
<artifactId>flink-connector-redis_${scala.binary.version}</artifactId>
<version>1.1-SNAPSHOT</version>
</dependency>
接下来就是定义一个类（或者静态内部类）实现RedisMapper即可，需要指定一个泛型，这里是Tuple2<String, Integer>，即写入到Redis中的数据的类型，并实现三个方法。第一个方法是getCommandDescription方法，返回RedisCommandDescription实例，在该构造方法中可以指定写入到Redis的方法类型为HSET，和Redis的additionalKey即value为HASH类型外面key的值；第二个方法getKeyFromData是指定value为HASH类型对应key的值；第三个方法geVauleFromData是指定value为HASH类型对应value的值。
在使用之前，先new FlinkJedisPoolConfig，设置Redis的ip地址或主机名、端口号、密码等。然后new RedisSink将准备好的conf和RedisWordCountMapper实例传入到其构造方法中，最后调用DataStream的addSink方法，将new好的RedisSink作为参数传入。
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.apache.flink.util.Collector;
/**
* 从指定的socket读取数据，对单词进行计算，将结果写入到Redis中
*/
public class RedisSinkDemo {
public static void main(String[] args) throws Exception {
//创建Flink流计算执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//创建DataStream
//Source
DataStreamSource<String> lines = env.socketTextStream("cs-28-86", 8888);
//调用Transformation开始
//调用Transformation
SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = lines.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
@Override
public void flatMap(String line, Collector<Tuple2<String, Integer>> collector) throws Exception {
String[] words = line.split(" ");
for (String word : words) {
//new Tuple2<String, Integer>(word, 1)
collector.collect(Tuple2.of(word, 1));
}
}
});
//分组
KeyedStream<Tuple2<String, Integer>, String> keyed = wordAndOne.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
@Override
public String getKey(Tuple2<String, Integer> tp) throws Exception {
return tp.f0;
}
});
//聚合
SingleOutputStreamOperator<Tuple2<String, Integer>> summed = keyed.sum(1);
//Transformation结束
//调用Sink
//summed.addSink()
FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("localhost").setDatabase(0).build();
summed.addSink(new RedisSink<Tuple2<String, Integer>>(conf, new RedisWordCountMapper()));
//启动执行
env.execute("StreamingWordCount");
}
public static class RedisWordCountMapper implements RedisMapper<Tuple2<String, Integer>> {
@Override
public RedisCommandDescription getCommandDescription() {
return new RedisCommandDescription(RedisCommand.HSET, "WORD_COUNT");
}
@Override
public String getKeyFromData(Tuple2<String, Integer> data) {
return data.f0;
}
@Override
public String getValueFromData(Tuple2<String, Integer> data) {
return data.f1.toString();
}
}
}1.1.7KafkaSink
在实际的生产环境中，经常会有一些场景，需要将Flink处理后的数据快速地写入到一个分布式、高吞吐、高可用、可用保证Exactly Once的消息中间件中，供其他的应用消费处理后的数据。Kafka就是Flink最好的黄金搭档，Flink不但可以从Kafka中消费数据，还可以将处理后的数据写入到Kafka，并且吞吐量高、数据安全、可以保证Exactly Once等。
Flink可以和Kafka多个版本整合，比如0.11.x、1.x、2.x等，从Flink1.9开始，使用的是kafka 2.2的客户端，所以这里使用kafka的版本是2.2.2，并且使用最新的API。
下面的例子就是将数据写入到Kafka中，首先要定义一个类实现KafkaSerializationSchema接口，指定一个泛型，String代表要写入到Kafka的数据为String类型。该类的功能是指定写入到Kafka中数据的序列化Schema，需要重写serialize方法，将要写入的数据转成二进制数组，并封装到一个ProducerRecord中返回。
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
public class KafkaSinkDemo {
public static void main(String[] args) throws Exception {
//local模式默认的并行度是当前机器的逻辑核的数量
Configuration configuration = new Configuration();
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
int parallelism0 = env.getParallelism();
System.out.println("执行环境默认的并行度：" + parallelism0);
DataStreamSource<String> lines = env.socketTextStream("cs-28-86", 8888);
//获取DataStream的并行度
int parallelism = lines.getParallelism();
System.out.println("SocketSource的并行度：" + parallelism);
//lines.writeAsText("file:///Users/xing/Desktop/out");
FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>(
"cs-28-87:9092,cs-28-88:9092,cs-28-89:9092", "wordcount18", new SimpleStringSchema()
);
lines.addSink(kafkaProducer);
env.execute();
}
}
启动nc –lk 8888 ,然后启动上述代码程序；
在nc窗口中输入数据，使用kafka可以消费到；
kafka消费wordcount18的topic：
[root@cs-28-88 ~]# kafka-console-consumer –zookeeper cs-28-88:2181 –topic wordcount18
然后将Kafka相关的参数设置到Properties中，再new FlinkKafkaProducer，将要写入的topic名称、Kafka序列化Schema、Properties和写入到Kafka的Semantic语义作为FlinkKafkaProducer构造方法参数传入。最好调用addSink方法将FlinkKafkaProducer的引用传入到该方法中。虽然下面的代码指定了EXACTLY_ONCE语义，但是没有开启Checkpointing，是没法实现的。具有怎样实现Exactly Once，会在后面原理深入的章节进行讲解。
1.1.8StreamFileDataSink
实时处理的数据，有一些场景要输出到其他分布式文件系统中，比如Hadoop HDFS、Amazon S3 （Simple Storage Service）、Aliyun OSS（Object Storage Service）等。因为这些分布式文件系统都具有高可用、可扩展、多副本、存储海量数据等特点。存储到分布式文件系统的数据，就可以做一些离线的数据分析，比如离线的数仓、数据挖掘、机器学习等。
从Flink 1.9开始，原来的Bucketing Sink已经标记为过时，在未来的版本将会被移除。推荐使用StreamFileDataSink，该Sink不但可以将数据写入到各种文件系统中，可以保证Exacly Once语义，还支持以列式存储的格式写入，功能更强大。
下面的例子是将数据写入到HDFS中，首先在maven项目的pom.xml文件引入HDFS文件系统的依赖：
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-filesystem_2.12</artifactId>
<version>1.12-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>2.6.0</version>
</dependency>
通过DefaultRollingPolicy这个工具类，指定文件滚动生成的策略。这里设置的文件滚动生成策略有两个，一个是距离上一次生成文件时间超过30秒，另一个是文件大小达到100 mb。这两个条件只要满足其中一个即可滚动生成文件。然后StreamingFileSink.forRowFormat方法将文件输出目录、文件写入的编码传入，再调用withRollingPolicy关联上面的文件滚动生成策略，接着调用build方法构建好StreamingFileSink，最后将其作为参数传入到addSink方法中。
1.1.9 JDBCSink
package com.bigdata.sink;
import com.bigdata.utils.DateUtil;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import java.util.Arrays;
import java.util.List;
/****
* @author songshiming
* @date 2022/12/10
* @desc
*/
public class SinkToMySql2 {
public static void main(String[] args) throws Exception {
System.out.println("start="+ DateUtil.getCurrentdatetime());
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// DataStreamSource<String> lineDataStreamSource = env.readTextFile("input/200.csv");
DataStreamSource<String> lineDataStreamSource = env.readTextFile("E://t1_trxrecord_20220821_V2.csv","gb2312");
SingleOutputStreamOperator<List<String>> wordList =
lineDataStreamSource.flatMap((String line, Collector<List<String>> out) -> {
// line = new String(line.getBytes("utf-8"),"gbk");
// System.out.println("line="+line);
String[] words = line.split(",");
out.collect(Arrays.asList(words));
}).returns(Types.LIST(Types.STRING));
String sqlStr ="INSERT INTO db_test.test_t1(TRXID, PARENT_TRXID, MERCHANT_NO, BRANCH_OFFICE, EXPAND_ORG, MAINTENANCE_ORG, STORE_CD, TERMINAL_NO, TRADE_TIME, SETTLEMENT_DATE, PRODUCT_NAME, TRADE_TP, TRADE_STA, TRADE_CARD_NO, TRADE_CARD_TP, ISSUER_CODE, TRADE_INIT_AMT, TRADE_AMT, BILLING_CYCLE, FEE_COLLECTION_STA, MER_FEE, SYSTEM_COST, BRAND_FEE, NET_PROFIT, MCC18, MCC42, ACCOUNT_ID, BUSINESS_TYPE, ORDER_NO, OTHER_MER_NO, OTHER_ACCOUNT_NO, SUBMIT_WAYS, TERMINAL_CODE, TERMINAL_BATCH, TERMINAL_TRACK_NO, TRADE_REFERENCE_NO, CHANNEL_NO, CHANNEL_MER_NO, TRADE_REMARKS, TRADE_ABSTRACT, TRADE_IP, CHANNEL_RET_CODE, SUBMIT_TIME, ERR_CODE, ERR_MSG, CHANNEL_TRADE_TP, INSTALLMENT_SUBSIDY_FEE, SUBSIDY_INFO, DCC_CURRENCY, DCC_AMT, DCC_EXCHANGE_RATE, PACKAGE_FEE, APP_ID, PACKAGE_ID) \n" +
"VALUES \n" +
"(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)" ;
wordList.addSink(JdbcSink.sink(
sqlStr,
((statement, word) -> {
for (int i = 0; i < word.size(); i++) {
String str = word.get(i).trim();
statement.setString(i+1, str );
}
}),
new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
.withUrl("jdbc:mysql://192.168.0.11:3306/db_test?useUnicode=true&characterEncoding=utf8")
.withDriverName("com.mysql.jdbc.Driver")
.withUsername("root")
.withPassword("123456")
.build()
));
env.execute();
System.out.println("start="+ DateUtil.getCurrentdatetime());
}
}
如果你向SCI期刊投过稿，那么你一定收到过来自审稿人或编辑的回信。
最近，笔者刚把一篇写得比较水的手稿投到了一本比较水的期刊，虽然写得不咋样，但是当时的笔者觉得这篇文章还不错。
过了一个月，笔者就收到审稿意见，里面有某位审稿人的“友好”评论：
I read this paper several times but do not understand it. I cannot comment on the paper. It makes no sense.（我读了几遍但没看懂。我无法评论这篇文章。它毫无意义。）
就这样，我的文章在一审中悲剧了，咱就是说，笔者当时真的脚趾抠出了三室一厅——也太尴尬（打击人）了吧！
不知道大家有没有类似的投稿经历：投稿前猛如虎，投稿后怂如狗！
图片来源：知乎@鸡糟的黄医桑
来自审稿人的毒舌
下面是一些网友们分享的“最狠”审稿意见，保证让你爆笑！
• “This paper is desperate. Please reject it completely and then block the author’s email ID so they can’t use the online system in future.”（这篇论文烂到家了。请彻底拒稿，然后封了作者的账号，免得他日后继续投稿。）
• “The writing and data presentation are so bad that I had to leave work and go home early and then spend time to wonder what life is about.”（写作水平和数据展示让我整个人都不好了，我不得不提前下班，早点回家，甚至开始怀疑人生。）
• “The biggest problem with this manu, which has nearly sucked the will to live out of me, is the terrible writing style.”（这篇稿件的写作风格太糟糕了，审得我都不想活了。）
• “This paper is awfully written. There is no adequate objective and no reasonable conclusion. The literature is quoted at random and not in the context of argument. I have doubts about the methods and whether the effort of data gathering is sufficient to arrive at a useful conclusion.”（这篇论文写得太烂了。目的不明确，结论不合理。引文都是乱引的，跟正文的论证完全对不上号。我怀疑这样的实验方法，也怀疑他们收集的数据不足以得出有效结论。）
• “The results are as weak as wet noodles.”（结论跟湿面条一样软弱无力。）
• “The authors are amateur.”（作者怕不是个兼职的。）
• “It is a bit for me that authors have used Python statistical analysis instead of SPSS or Matlab as usual.Please,explain.”（本文用了 Python 而不是像通常用 SPSS 和 Matlab 处理数据，为啥？）
• “Did you have seizure while writing this sentence?Because I fell like I had one while reading it.”（你写文章的时候犯癫痫了吗？反正我觉得我读的时候犯了。）
• “Why chase a gene in this ridiculous organism?”（为什么在这种荒诞的生物里追求一个基因。）
• “This paper introduce tools to answer questions which it does not seem many people are interested in.”（本文介绍了些解决问题的工具，然而人们并不感兴趣。）
• “The English language ranks this manuscript among the top 5 worst manuscripts I have ever reviewed.”（这篇文章的英语水平可以排进我看过的最烂的前五。）
• “I started reading this manuscript with much anticipation.But my enthusiasm was short lived.”（我满怀期待读这篇文章，但热情很短。）
• “I have rarely a more blown-up and annoy paper in the last couple of years than this hot-air balloon manuscript.”（我过去几年都没读过这篇像屁一样烦人的文章。）
• “Figure3.This figure is silly.”（这图好蠢呐。）
• “This paper makes no contribution.”（这文章有锤子贡献。）
• “But now, there are over 1000 articles on this topic. But this author have not read a single.”（这个方向至少有一千篇文章，作者没读过一篇。）
• “Reject-More holes than my grandads’ string vest.”（拒了，漏洞比我爷爷的背心上的网眼还多。）
• “The biggest problem with manuscript, which has nearly sucked the will to live out of me, is the terrible writing style.”（这篇稿子审得我都想自挂东南枝了。）
• “This paper is desperate. Please reject it complete and then block The author’s email ID so they can’t use The online system future.”（文章太差了，麻烦封掉作者的账号以免他以后再投稿。）
• “The writing and data presentation are so bad that I had to leave and go home early and then spend time to wonder what life is about.”（手稿和数据显示让我整个人都不好了，不得不提前下班，开始怀疑人生。）
• “The presentation is of a standard that I would reject from an undergraduate student.”（就算回到本科时期，我也会拒掉这种文章。）
• “This paper is, simply manure.”（这篇论文，简直就是肥料。）
•“What the authors have done is an insult to science.”（作者所做的是对科学的侮辱。）
• “You should look closely at a career outside of science.”（你应该仔细考虑一下科学以外的职业。）
所以，想想如果你平时投稿时只是收到一些普通的拒稿邮件，哪怕就文章内容批评几句，已经是多么幸运的事！
文章怎么写才不用惨遭审稿人“毒舌”？使用恰当的学术语言，掌握写作技巧才是正经事。比如这个审稿意见“这篇文章的英语水平可以排进我看过的最烂的前五”，可见审稿人对语言的要求是非常高的！
另外，实在没空润色文章，作者们还可以体验一下投必得的语言润色，助你免遭审稿人吐槽的同时，增加论文发表的成功率。
来自审稿人的称赞
见识过了审稿人的毒舌和幽默，再来见识一下审稿人怎么夸人。但当审稿人遇到非常棒的论文，也会给出衷心的好评。
• “Very much enjoyed reading this one, and do not have any significant comments. Wish I had thought of this one.”（这篇论文读起来简直是一种享受，没有任何意见。这么好的点子我当初怎么没想到！）
• “There was little I could think of to improve this nice paper.”（这篇论文好得我都不知道要怎么改了。）
• “This is as solid a write up as I have seen, many spend much more time and space to say considerably less. It is a perfect example of a compact report.”（这是我见过的最确凿的报告，许多人花更多时间精力写出来的比这个差多了。作为一篇翔实的报告，这是一个完美的例子。）
• “landmark paper on P. putida physiology.”（假单胞菌生理学研究的里程碑。）
• “It is always a joy to review manus such as this. Well conceived, well executed, well edited. Clean. Pristine. From start to finish.”（评审这样的稿件是件人生乐事：构思周详，实施得当，写得漂亮。干净，纯朴，从头至尾。）
欢迎大家留言分享让您记忆深刻的审稿意见！
最新论文发表于Q1期刊
这是一篇由投必得助力 Hubei Academy of Agricultural Sciences，Wuhan Lichen Biotechnology Co， Wuhan University的老师和学生们发表于Journal of Plant Physiology期刊的SCI论文。下面是论文标题！
论文标题
论文下载地址：
https://doi.org/10.1016/j.jplph.2022.153688
有需要的读者可以自行下载！
这篇论文2021年12月2日提交，已经于2022年3月25日接受，2022年4月14日线上见刊，耗时近5个月。这篇论文本身具有很高的研究价值和重大的研究成果，又有投必得专业论文编辑为写作质量保驾护航，成功发表属于意料之中！
• Received2December2021,
• Revised24March2022,
• Accepted25March2022,
• Availableonline14April2022,
• VersionofRecord21April2022.
这篇论文的Acknowledgements部分，作者对投必得润色编辑服务予以致谢！我们由衷感谢作者团队对我们的信任和支持！
论文致谢
详情请看：
投必得提供了“投稿期刊推荐”服务，旨在帮助您更好地了解自己稿件的水平，划定合适的投稿期刊范围，从而让您在期刊选择时避免走不必要的弯路，缩短稿件被接受的周期。编辑会根据您的文章先进行编辑前评审评估，总结成一份预评估报告（Pre-evaluation report）；并按照您对于期刊的要求，推荐 3 个目标期刊，填写一份期刊推荐表。
iJournal期刊查询
接下来我们用投必得的 iJournal 期刊查询与选择平台
（
https://ijournal.topeditsci.com/home）来查看一下这本期刊的详细信息。
Journal of Plant Physiology是一本广谱期刊，欢迎植物生理学所有主要领域的高质量投稿，包括植物生物化学、功能生物技术、计算和合成植物生物学、生长和发育、光合作用和呼吸作用、运输和易位，植物-微生物相互作用、生物和非生物胁迫。从分子和细胞到有机体及其环境，所有整合水平的研究都受到欢迎，并有望使用最先进的方法。纯基因表达研究不在我们期刊的重点范围内。要考虑发表，论文必须对生理过程的机制理解做出重大贡献，而不仅仅是描述性的，或证实以前的结果。我们鼓励提交探索非模型以及公认模型物种的生理学以及连接基础研究和应用研究的论文。例如，对显示出提高农业效率的新生理机制的农业植物的研究是受欢迎的。在不受控制的情况下进行的研究（例如现场条件）不提供机械洞察力将不被考虑发表。（以上机器翻译，仅供参考）
ISSN：0176-1617
出版商：ELSEVIERGMBH
影响因子：3.549（2020年）
期刊官网：
JournalofPlantPhysiology|ScienceDirect.combyElsevier
影响因子
分区截图
年发文量
iJournal 提供了期刊详情、基础版、升级版中科院 JCR 分区、科睿唯安的 JCR 分区、最新影响因子（IF）、引用因子、高引用文章、发文量等信息，有助于您的选刊。此外，投必得提供投稿期刊推荐服务，欢迎大家咨询！
写在最后
以上分析，仅为一家之言，如有不当，敬请指出。仅供参考~