android socket 推送服务版本

你的名字 2023-10-10 11:39 45阅读 0赞

客户端代码：

主功能界面;

package com.rf.pushclient;
    
    import android.os.Bundle;
    import android.app.Activity;
    import android.content.Intent;
    import android.view.Menu;
    import android.view.View;
    import android.view.View.OnClickListener;
    import android.widget.Button;
    
    public class MainActivity extends Activity {
    	private Button mConnect;
    	private Button mDisConnect;
    	@Override
    	protected void onCreate(Bundle savedInstanceState) {
    		super.onCreate(savedInstanceState);
    		setContentView(R.layout.activity_main);
    		
    		mConnect=(Button) findViewById(R.id.connect);
    		mDisConnect=(Button) findViewById(R.id.disconnect);
    		mConnect.setOnClickListener(new OnClickListener(){
    
    			@Override
    			public void onClick(View v) {
    				// TODO Auto-generated method stub
    				startService(new Intent(MainActivity.this, PushService.class));
    			}
    			
    		});
    		mDisConnect.setOnClickListener(new OnClickListener(){
    
    			@Override
    			public void onClick(View v) {
    				// TODO Auto-generated method stub
    				stopService(new Intent(MainActivity.this, PushService.class));
    			}
    			
    		});
    		
    	}
    
    
    
    }

服务类：

package com.rf.pushclient;
    
    import java.net.InetSocketAddress;
    
    import android.app.Notification;
    import android.app.NotificationManager;
    import android.app.PendingIntent;
    import android.app.Service;
    import android.content.Intent;
    import android.os.Handler;
    import android.os.IBinder;
    import android.os.Message;
    
    public class PushService extends Service {
    	private PushClient mClient;
    
    	private Handler mHandler;
    
    	@Override
    	public IBinder onBind(Intent intent) {
    		return null;
    	}
    
    	@Override
    	public void onCreate() {
    		super.onCreate();
    		//加载Handler
    		initHandler();
    	}
    
    	@Override
    	public int onStartCommand(Intent intent, int flags, int startId) {
    		//启动连接通道线程
    		mClient = new PushClient(new InetSocketAddress("192.168.10.111", 10001),
    				mHandler);
    		//线程启动
    		mClient.start();
    		return super.onStartCommand(intent, flags, startId);
    	}
    
    	@Override
    	public void onDestroy() {
    	   //线程销毁
    		if (mClient != null) {
    			mClient.disConnect();
    		}
    		super.onDestroy();
    	}
    
    	/**
    	 * 初始化Handler
    	 */
    	private void initHandler() {
    		mHandler=new Handler(){
    
    			@Override
    			public void handleMessage(Message msg) {
    				// TODO Auto-generated method stub
    				switch(msg.what){
    				case 100:
    					String message=msg.obj.toString();
    					//状态栏提示相关信息
    					showNotification(message);
    					break;
    				default:
    					break;
    				
    				}
    			}
    			
    		};
    	}
    
    	/**
    	 * 
    	 * 在状态栏显示通知
    	 */
    
    	@SuppressWarnings("deprecation")
    	private void showNotification(String msg) {
    		// 创建一个NotificationManager的引用
    		NotificationManager notificationManager = (NotificationManager) getSystemService(android.content.Context.NOTIFICATION_SERVICE);
    
    		// 定义Notification的各种属性
    		Notification notification = new Notification(R.drawable.ic_launcher,
    				msg, System.currentTimeMillis());
    
    		// FLAG_AUTO_CANCEL 该通知能被状态栏的清除按钮给清除掉
    		// FLAG_NO_CLEAR 该通知不能被状态栏的清除按钮给清除掉
    		// FLAG_ONGOING_EVENT 通知放置在正在运行
    		// FLAG_INSISTENT 是否一直进行，比如音乐一直播放，知道用户响应
    		notification.flags |= Notification.FLAG_AUTO_CANCEL; // 表明在点击了通知栏中的"清除通知"后，此通知不清除，经常与FLAG_ONGOING_EVENT一起使用
    
    		// DEFAULT_ALL 使用所有默认值，比如声音，震动，闪屏等等
    		// DEFAULT_LIGHTS 使用默认闪光提示
    		// DEFAULT_SOUND 使用默认提示声音
    		// DEFAULT_VIBRATE 使用默认手机震动，需加上<uses-permission
    		// android:name="android.permission.VIBRATE" />权限
    		notification.defaults = Notification.DEFAULT_SOUND;
    
    		// 设置通知的事件消息
    		CharSequence contentTitle = msg; // 通知栏标题
    		CharSequence contentText = msg; // 通知栏内容
    		Intent notificationIntent = new Intent(this, TestActivity.class); // 点击该通知后要跳转的Activity
    
    		PendingIntent contentItent = PendingIntent.getActivity(this, 0,
    				notificationIntent, 0);
    		notification.setLatestEventInfo(this, contentTitle, contentText,
    				contentItent);
    		// 把Notification传递给NotificationManager
    		notificationManager.notify(0, notification);
    	}
    
    }

线程类：

/*
     * 
     */
    package com.rf.pushclient;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.SocketChannel;
    import java.util.Iterator;
    import java.util.concurrent.atomic.AtomicBoolean;
    import android.os.Handler;
    import android.os.Message;
    
    
    
    /**
     * 
     * @author zengjiantao
     * @date 2013-4-8
     */
    public class PushClient extends Thread {
    
    	private static final int BUFFER_SIZE = 1024;
    
    	/**
    	 * 远程地址
    	 */
    	private final InetSocketAddress mRemoteAddress;
    
    	/**
    	 * 连接通道
    	 */
    	private SocketChannel mSocketChannel;
    
    	/**
    	 * 接收缓冲区
    	 */
    	private final ByteBuffer mReceiveBuf;
    
    	/**
    	 * 端口选择器
    	 */
    	private Selector mSelector;
    
    	/**
    	 * 线程是否结束的标志
    	 */
    	private final AtomicBoolean mShutdown;
    	
    	/**
    	 *  消息处理
    	 */
    	private final Handler mHandler;
    
    	static {
    		java.lang.System.setProperty("java.net.preferIPv4Stack", "true");
    		java.lang.System.setProperty("java.net.preferIPv6Addresses", "false");
    	}
    
    	public PushClient(InetSocketAddress remoteAddress, Handler handler) {
    		mRemoteAddress = remoteAddress;
    		mHandler = handler;
    
    		// 初始化缓冲区
    		mReceiveBuf = ByteBuffer.allocateDirect(BUFFER_SIZE);
    		if (mSelector == null) {
    			// 创建新的Selector
    			try {
    				mSelector = Selector.open();
    			} catch (final IOException e) {
    				e.printStackTrace();
    			}
    		}
    		mShutdown = new AtomicBoolean(false);
    	}
    
    	/**
    	 * 打开通道
    	 */
    	private void startup() {
    		try {
    			// 打开通道
    			mSocketChannel = SocketChannel.open();
    			// 绑定到本地端口
    			mSocketChannel.socket().setSoTimeout(30000);
    			mSocketChannel.configureBlocking(false);
    			if (mSocketChannel.connect(mRemoteAddress)) {
    				System.out.println("开始建立连接:" + mRemoteAddress);
    			}
    			mSocketChannel.register(mSelector, SelectionKey.OP_CONNECT
    					| SelectionKey.OP_READ, this);
    			System.out.println("端口打开成功");
    
    		} catch (final IOException e1) {
    			e1.printStackTrace();
    		}
    	}
    
    	private void select() {
    		int nums = 0;
    		try {
    			if (mSelector == null) {
    				return;
    			}
    			nums = mSelector.select(1000);
    		} catch (final Exception e) {
    			e.printStackTrace();
    		}
    
    		// 如果select返回大于0，处理事件
    		if (nums > 0) {
    			Iterator<SelectionKey> iterator = mSelector.selectedKeys()
    					.iterator();
    			while (iterator.hasNext()) {
    				// 得到下一个Key
    				final SelectionKey key = iterator.next();
    				iterator.remove();
    				// 检查其是否还有效
    				if (!key.isValid()) {
    					continue;
    				}
    				// 处理事件
    				try {
    					if (key.isConnectable()) {
    						connect();
    					} else if (key.isReadable()) {
    						read(key);
    					}
    				} catch (final Exception e) {
    					e.printStackTrace();
    					key.cancel();
    				}
    			}
    		}
    	}
    
    	@Override
    	public void run() {
    		//执行第一个方法
    		startup();
    		// 启动主循环流程
    		while (!mShutdown.get()) {
    			try {
    				// do select
    				select();
    				try {
    					Thread.sleep(1000);
    				} catch (final Exception e) {
    					e.printStackTrace();
    				}
    			} catch (final Exception e) {
    				e.printStackTrace();
    			}
    		}
    		shutdown();
    	}
    
    	private void connect() throws IOException {
    		if (isConnected()) {
    			return;
    		}
    		// 完成SocketChannel的连接
    		mSocketChannel.finishConnect();
    		while (!mSocketChannel.isConnected()) {
    			try {
    				Thread.sleep(300);
    			} catch (final InterruptedException e) {
    				e.printStackTrace();
    			}
    			mSocketChannel.finishConnect();
    		}
    
    	}
    
    	public void disConnect() {
    		mShutdown.set(true);
    	}
    
    	private void shutdown() {
    		if (isConnected()) {
    			try {
    				mSocketChannel.close();
    				while (mSocketChannel.isOpen()) {
    					try {
    						Thread.sleep(300);
    					} catch (final InterruptedException e) {
    						e.printStackTrace();
    					}
    					mSocketChannel.close();
    				}
    				System.out.println("端口关闭成功");
    			} catch (final IOException e) {
    				System.err.println("端口关闭错误:");
    				e.printStackTrace();
    			} finally {
    				mSocketChannel = null;
    			}
    		} else {
    			System.out.println("通道为空或者没有连接");
    		}
    		// 关闭端口选择器
    		if (mSelector != null) {
    			try {
    				mSelector.close();
    				System.out.println("端口选择器关闭成功");
    			} catch (IOException e) {
    				e.printStackTrace();
    			} finally {
    				mSelector = null;
    			}
    		}
    	}
    
    	private void read(SelectionKey key) throws IOException {
    		// 接收消息
    		final byte[] msg = recieve();
    		if (msg != null) {
    			String tmp = new String(msg);
    			System.out.println("返回内容：");
    			System.out.println(tmp);
    			if (mHandler != null) {
    				Message message = mHandler.obtainMessage(100);
    				message.obj = tmp;
    				mHandler.sendMessage(message);
    			}
    		}
    	}
    
    	private byte[] recieve() throws IOException {
    		if (isConnected()) {
    			int len = 0;
    			int readBytes = 0;
    
    			synchronized (mReceiveBuf) {
    				mReceiveBuf.clear();
    				try {
    					while ((len = mSocketChannel.read(mReceiveBuf)) > 0) {
    						readBytes += len;
    					}
    				} finally {
    					mReceiveBuf.flip();
    				}
    				if (readBytes > 0) {
    					final byte[] tmp = new byte[readBytes];
    					mReceiveBuf.get(tmp);
    					return tmp;
    				} else {
    					System.out.println("接收到数据为空,重新启动连接");
    					return null;
    				}
    			}
    		} else {
    			System.out.println("端口没有连接");
    		}
    		return null;
    	}
    
    	private boolean isConnected() {
    		return mSocketChannel != null && mSocketChannel.isConnected();
    	}
    }

服务端代码：

package com.rf.pushserver;
    
    import java.io.BufferedReader;
    import java.io.IOException;
    import java.io.InputStreamReader;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.ServerSocketChannel;
    import java.nio.channels.SocketChannel;
    import java.util.Iterator;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.atomic.AtomicBoolean;
    
    /**
     * 消息推送服务器
     * 
     * @author zengjiantao
     * @date 2013-4-8
     */
    public class PushServer extends Thread {
    
    	private static final int BUFFER_SIZE = 1024;
    
    	/**
    	 * 服务器连接通道
    	 */
    	private ServerSocketChannel serverSocketChannel;
    
    	/**
    	 * 发送缓冲区
    	 */
    	private final ByteBuffer sendBuf;
    
    	/**
    	 * 端口选择器
    	 */
    	private Selector selector;
    
    	/**
    	 * 服务器端口
    	 */
    	private final int mPort;
    
    	/**
    	 * 线程是否结束的标志
    	 */
    	private final AtomicBoolean shutdown;
    
    	/**
    	 * 发送消息的开关
    	 */
    	private final AtomicBoolean sendable;
    
    	/**
    	 * 发送消息的内容
    	 */
    	private String sendMsg;
    
    	private final ExecutorService executorService;
    
    	public PushServer(int port) {
    		mPort = port;
    		// 初始化缓冲区
    		sendBuf = ByteBuffer.allocateDirect(BUFFER_SIZE);
    		if (selector == null) {
    			// 创建新的Selector
    			try {
    				selector = Selector.open();
    			} catch (final IOException e) {
    				e.printStackTrace();
    			}
    		}
    
    		startup();
    		executorService = Executors.newFixedThreadPool(10);
    		shutdown = new AtomicBoolean(false);
    		sendable = new AtomicBoolean(false);
    	}
    
    	private void startup() {
    		try {
    			// 打开通道
    			serverSocketChannel = ServerSocketChannel.open();
    			// 绑定到本地端口
    			serverSocketChannel.socket().setSoTimeout(30000);
    			serverSocketChannel.configureBlocking(false);
    			serverSocketChannel.socket().bind(new InetSocketAddress(mPort));
    			serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
    			System.out.println("服务器端口打开成功");
    
    		} catch (final IOException e1) {
    			e1.printStackTrace();
    		}
    	}
    
    	private void select() {
    		int nums = 0;
    		try {
    			if (selector == null) {
    				return;
    			}
    			nums = selector.select(1000L);
    		} catch (final Exception e) {
    			e.printStackTrace();
    		}
    
    		// 如果select返回大于0，处理事件
    		if (nums > 0) {
    			Iterator<SelectionKey> iterator = selector.selectedKeys()
    					.iterator();
    			while (iterator.hasNext()) {
    				// 得到下一个Key
    				final SelectionKey key = iterator.next();
    				iterator.remove();
    				// 检查其是否还有效
    				if (!key.isValid()) {
    					continue;
    				}
    
    				// 处理事件
    				if (key.isAcceptable()) {
    					executorService.execute(new Accepter(key));
    					// accept(key);
    				} else if (key.isWritable()) {
    					if (sendable.get()) {
    						executorService.execute(new Sender(key, sendMsg));
    					}
    				}
    			}
    			if (sendable.get()) {
    				System.out.println("结束推送消息了");
    			}
    			sendable.set(false);
    		}
    	}
    
    	/**
    	 * 用于连接的Runnable
    	 * 
    	 * @author zengjiantao
    	 * @date 2013-4-11
    	 */
    	class Accepter implements Runnable {
    
    		private final SelectionKey key;
    
    		public Accepter(SelectionKey key) {
    			this.key = key;
    		}
    
    		public void run() {
    			accept(key);
    		}
    
    	}
    
    	/**
    	 * 用于发送消息的Runnable
    	 * 
    	 * @author zengjiantao
    	 * @date 2013-4-11
    	 */
    	class Sender implements Runnable {
    
    		private final SelectionKey key;
    
    		private final String msg;
    
    		public Sender(SelectionKey key, String msg) {
    			this.key = key;
    			this.msg = msg;
    		}
    
    		public void run() {
    			send(key, msg);
    		}
    
    	}
    
    	/**
    	 * 接收客户端
    	 * 
    	 * @param key
    	 * @throws IOException
    	 */
    	private void accept(SelectionKey key) {
    		// 打开通道
    		try {
    			SocketChannel socketChannel = ((ServerSocketChannel) key.channel())
    					.accept();
    			// 绑定到本地端口
    			socketChannel.socket().setSoTimeout(30000);
    			socketChannel.configureBlocking(false);
    			synchronized (selector) {
    				socketChannel.register(selector, SelectionKey.OP_WRITE, this);
    			}
    			System.out.println("端口打开成功");
    		} catch (IOException e) {
    			System.out.println("端口打开失败");
    			e.printStackTrace();
    			key.cancel();
    		}
    	}
    
    	@Override
    	public void run() {
    		// 启动主循环流程
    		while (!shutdown.get()) {
    			try {
    				select();
    				try {
    					Thread.sleep(1000L);
    				} catch (final Exception e) {
    					e.printStackTrace();
    				}
    			} catch (final Exception e) {
    				e.printStackTrace();
    			}
    		}
    		shutdown();
    	}
    
    	/**
    	 * 打开发送消息的开关
    	 * 
    	 * @param msg
    	 */
    	void send(final String msg) {
    		sendMsg = msg;
    		sendable.set(true);
    		System.out.println("开始推送消息了");
    	}
    
    	/**
    	 * 向指定连接发送消息
    	 * 
    	 * @param key
    	 * @param msg
    	 */
    	private void send(final SelectionKey key, final String msg) {
    		try {
    			byte[] out = msg.getBytes();
    			if (out == null || out.length < 1) {
    				return;
    			}
    			synchronized (sendBuf) {
    				sendBuf.clear();
    				sendBuf.put(out);
    				sendBuf.flip();
    			}
    			SocketChannel socketChannel = (SocketChannel) key.channel();
    			socketChannel.write(sendBuf);
    		} catch (final IOException e) {
    			e.printStackTrace();
    		}
    	}
    
    	/**
    	 * 断开连接
    	 */
    	public void disConnect() {
    		shutdown.set(true);
    	}
    
    	/**
    	 * 关闭端口选择器
    	 */
    	private void shutdown() {
    		if (serverSocketChannel != null) {
    			try {
    				serverSocketChannel.close();
    				while (serverSocketChannel.isOpen()) {
    					try {
    						Thread.sleep(300L);
    					} catch (final InterruptedException e) {
    						e.printStackTrace();
    					}
    					serverSocketChannel.close();
    				}
    				System.out.println("端口关闭成功");
    			} catch (IOException e1) {
    				System.err.println("端口关闭错误:");
    				e1.printStackTrace();
    			} finally {
    				serverSocketChannel = null;
    			}
    		}
    		// 关闭端口选择器
    		if (selector != null) {
    			try {
    				selector.close();
    				System.out.println("端口选择器关闭成功");
    			} catch (IOException e) {
    				e.printStackTrace();
    			} finally {
    				selector = null;
    			}
    		}
    	}
    
    	public static void main(String[] args) {
    		try {
    			final PushServer server = new PushServer(10001);
    			server.start();
    			new Thread(new Runnable() {
    
    				public void run() {										
    						
    					try {
    						InputStreamReader input = new InputStreamReader(
    								System.in);
    						BufferedReader br = new BufferedReader(input);
    						String sendText = br.readLine();
    						server.send(sendText);
    					} catch (IOException e) {
    						e.printStackTrace();
    					}
    
    						
    
    				
    				}
    			}).start();
    
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    	}
    }