基于Flume做FTP文件实时同步到本地磁盘的windows服务。

package com.syher.flume;

import com.google.common.collect.Lists;
import com.urey.flume.sink.taildir.SafeRollingFileSink;
import org.apache.flume.*;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.channel.MemoryChannel;
import org.apache.flume.channel.ReplicatingChannelSelector;
import org.apache.flume.conf.Configurables;
import org.apache.flume.sink.DefaultSinkProcessor;
import org.apache.flume.sink.RollingFileSink;
import org.apache.flume.source.PollableSourceRunner;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.keedio.flume.source.ftp.source.Source;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

//@RunWith(SpringRunner.class)
//@SpringBootTest
public class SpringBootFlumeApplicationTests {
    Context defaultContext = new Context();

    @Before
    public void init() throws Exception {
        Map prop = new HashMap();
        prop.put("channel.capacity", "1000");
        prop.put("channel.transactionCapacity", "1000");

        prop.put("source.client.source", "ftp");
        prop.put("source.name.server", "192.168.1.150");
        prop.put("source.port", "21");
        prop.put("source.user", "username");
        prop.put("source.password", "secret");
        prop.put("source.working.directory", "/ftp/source");
        prop.put("source.filter.pattern", ".+\\.pdf");
       // prop.put("source.folder", "/ftp");
        prop.put("source.flushlines", "false");

        prop.put("sink.sink.directory", "G:/ftp/target/rolling");
        prop.put("sink.sink.moveFile", "false");
        prop.put("sink.sink.targetDirectory", "G:/ftp/target/PDFfiles");
        prop.put("sink.sink.useCopy", "true");
        prop.put("sink.sink.copyDirectory", "G:/ftp/target/copy");
        prop.put("sink.sink.useFileSuffix", "false");
        prop.put("sink.sink.fileSuffix", ".log");
        defaultContext.putAll(prop);
    }

    public MemoryChannel getChannel() {
        MemoryChannel channel = new MemoryChannel();
        channel.setName("channel");

        configure(channel, "channel.");
        return channel;
    }

    public Source getSource(Channel channel) {
        Source source = new Source();
        source.setName("source");

        ChannelSelector selector = new ReplicatingChannelSelector();
        selector.setChannels(Lists.newArrayList(channel));

        ChannelProcessor processor = new ChannelProcessor(selector);
        source.setChannelProcessor(processor);

        configure(source, "source.");
        return source;
    }

    public Sink getSink(Channel channel) {
        SafeRollingFileSink sink = new SafeRollingFileSink();
        sink.setName("sink");

        sink.setChannel(channel);

        configure(sink, "sink.");
        return sink;
    }

    public void configure(Object target, String prefixProperty) {
        Context context = new Context();
        context.putAll(defaultContext.getSubProperties(prefixProperty));
        Configurables.configure(target, context);
    }

    @Test
    public void contextLoads() throws Exception {
        MemoryChannel channel = getChannel();
        Source source = getSource(channel);
        Sink sink = getSink(channel);

        PollableSourceRunner sourceRunner = new PollableSourceRunner();
        sourceRunner.setSource(source);
        channel.start();
        sourceRunner.start();

        SinkProcessor sinkProcessor = new DefaultSinkProcessor();
        sinkProcessor.setSinks(Arrays.asList(sink));
        SinkRunner sinkRunner = new SinkRunner(sinkProcessor);

        channel.start();
        sourceRunner.start();
        sinkRunner.start();

        while (!Thread.interrupted()) {
            Thread.sleep(200);
        }
    }

}