Flume 高可用 负载均衡问题

首先写一个普通的Java类
这个类主要就是向hostName iP的port节点发送信息

package flumeTest;

import java.nio.charset.Charset;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import org.apache.flume.Event;
import org.apache.flume.api.RpcClient;
import org.apache.flume.api.RpcClientFactory;
import org.apache.flume.event.EventBuilder;

//链接avro的flume source 发送event 到flume agent
public class FlumeClient {
    private RpcClient flumeClient;
    private String hostName;
    private int port;
    
    public FlumeClient(String hostname,int port){
        this.hostName =hostname;
        this.port=port;
        this.flumeClient=RpcClientFactory.getDefaultInstance(hostname, port);
    }
    //把字符串消息发送event到avro source
    public void sendEvent(String msg){
        Map<String, String> headers =new HashMap<String, String>();
        headers.put("timestamp", String.valueOf(new Date().getTime()));
        //构建event
        Event event =EventBuilder.withBody(msg, Charset.forName("UTF-8"), headers);
        
        try{
            flumeClient.append(event);
        }catch (Exception e) {
            e.printStackTrace();
            flumeClient.close();
            flumeClient=null;
            flumeClient=RpcClientFactory.getDefaultInstance(hostName, port);
        }
    }
    public void close(){
        flumeClient.close();
        
    }
    
    //这个类的作用就是向hostName的port端口输入Flume定义的RpcClient avro格式的内容
    public static void main(String[] args) {
        FlumeClient flumeClient =new FlumeClient("master", 5555);
        String bMsg="fromjava-msg";
        for(int i=0;i<100;i++){
            flumeClient.sendEvent(bMsg+i);
        }
        flumeClient.close();
    }
}

conf文件
不在展示 很简单 source为avro sink为logger channel为memory

高可用的方式

package flumeTest;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.Charset;
import java.util.Properties;

import org.apache.flume.Event;
import org.apache.flume.api.RpcClient;
import org.apache.flume.api.RpcClientFactory;
import org.apache.flume.event.EventBuilder;

//这个类主要是保证了Flume的高可用 通过properties文件定义Client类型是default_failover
//
public class FailoverClient {
    private Properties properties;
    private RpcClient failoverClient;
    
    //初始化rpcclient
    public FailoverClient() throws IOException{
        this.properties=new Properties();
        InputStream inputStream = new FileInputStream("E:\\new workspace\\flumeTest\\src\\main\\resources\\failover_client.conf");
        properties.load(inputStream);
        this.failoverClient=RpcClientFactory.getInstance(properties);
    }
    //发送消息
    public void sendEvent(String msg){
        
        Event event =EventBuilder.withBody(msg,Charset.forName("UTF-8"));
        try {
            failoverClient.append(event);
        } catch (Exception e) {
        e.printStackTrace();
        
        }
    }
    
    public void close(){
        failoverClient.close();
    }
    public static void main(String[] args) throws IOException, InterruptedException{
        FailoverClient failoverClient =new FailoverClient();
        String msg ="message_";
        for(int i=1;i<100;i++){
            
            failoverClient.sendEvent(msg+i);
            Thread.sleep(1000);
        }
        failoverClient.close();
    }
}

properties文件

client.type=default_failover

hosts=h1 h2

hosts.h1=master:8888
hosts.h2=slave1:8888

max-attempts=3

负载均衡的Java类和高可用的类几乎没有什么改变 变得只是properties文件

package flumeTest;

import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.Properties;

import org.apache.flume.Event;
import org.apache.flume.api.RpcClient;
import org.apache.flume.api.RpcClientFactory;
import org.apache.flume.event.EventBuilder;

public class LoadBalanceClient {
    private RpcClient lbClient;
    private Properties properties;
    
    public LoadBalanceClient() throws FileNotFoundException, IOException{
        
        this.properties =new Properties();
        properties.load(new FileInputStream("src/main/resources/load_balance.conf"));
        this.lbClient =RpcClientFactory.getInstance(properties);
    }
    public void sentEvent(String msg){
        Event event =EventBuilder.withBody(msg, Charset.forName("UTF-8"));
        try {
            lbClient.append(event);
        } catch (Exception e) {
        e.printStackTrace();
        }
    }
    public void close(){
        lbClient.close();
    }
    
    public static void main(String[] args) throws FileNotFoundException, IOException {
        LoadBalanceClient loadBalanceClient =new LoadBalanceClient();
        String msg ="msg_";
        for (int i = 0; i < 100; i++) {
            loadBalanceClient.sentEvent(msg+i);
        }
        loadBalanceClient.close();
        
    }
}

properties

client.type=default_loadbalance

hosts=h1 h2

hosts.h1=master:8888
hosts.h2=slave1:8888

host-selector-random